gRPC Connector

Call a gRPC Service from MQTT

A gRPC connector opens a channel from the broker to an external gRPC service and maps MQTT topics onto that service’s calls. Publish bytes on a source topic and the connector sends them; replies are published back onto a destination topic.

This page is the gRPC client connector (DEFINE ROUTE … WITH TYPE GRPC). The broker connects out to a gRPC service. You do not call it from an Action — there is no CALL GRPC.To let an AI client operate this broker, use the MCP Server. The hosted documentation MCP searches docs.coreflux.org and does not open a gRPC channel.

Like a phone line to another building. The connector keeps one line open to the service (the channel). Each conversation on that line is a call: MQTT bytes go in, and the reply bytes come back on another topic.

gRPC connectors are beta. Coreflux HUB lists the gRPC template with the other streaming connectors when the broker exposes it. The fields on that template are the same GRPC_CONFIG parameters as the LoT on this page.

When to use it

A service you already run speaks gRPC, and MQTT messages should invoke it without a separate gateway.
Each device needs its own long-lived stream. With the default session scope, one IMEI (or any other id in the source topic) gets its own bidirectional call.
The exchange is one request and one reply, a server stream, a client stream, or a bidirectional stream.

Use a REST API connector for HTTP. Use the Kafka connector for an event log. Encode and decode protobuf with LoT models — this connector forwards raw bytes and does not read a .proto file.

Define the channel once, then map the MQTT topics that feed it.

Quick Start

Unary call
Server stream
One call per device

One MQTT message opens one call and publishes the single reply:

DEFINE ROUTE EchoGrpc WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS "grpc.example.com"
        WITH PORT '50051'
        WITH USE_TLS "false"
        WITH SERVICE "echo.Echo"
        WITH METHOD "Unary"
        WITH RPC_TYPE "UNARY"
    ADD EVENT CallEcho
        WITH SOURCE_TOPIC "grpc/echo/in"
        WITH DESTINATION_TOPIC "grpc/echo/out"

A payload on grpc/echo/in is the request. The response bytes are published to grpc/echo/out.

One request, then every message the server sends until it finishes:

DEFINE ROUTE TickStream WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '50051'
        WITH USE_TLS "false"
        WITH SERVICE "echo.Echo"
        WITH METHOD "ServerStream"
        WITH RPC_TYPE "SERVER_STREAM"
    ADD EVENT Tick
        WITH SOURCE_TOPIC "grpc/ticks/request"
        WITH DESTINATION_TOPIC "grpc/ticks/data"

Each server message is published to grpc/ticks/data as it arrives.

A bidirectional call per device topic. The id in the topic — an IMEI, a serial number, a session id — is the call key:

DEFINE ROUTE DeviceBidi WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '443'
        WITH USE_TLS "true"
        WITH SERVICE "echo.Echo"
        WITH METHOD "Bidi"
        WITH RPC_TYPE "BIDI"
        WITH SESSION_SCOPE "TOPIC"
        WITH DEADLINE '300'
        WITH IDLE_TIMEOUT '15'
    ADD EVENT Converse
        WITH SOURCE_TOPIC "devices/+/uplink"
        WITH DESTINATION_TOPIC "devices/{topic.2}/downlink"

devices/490154203237518/uplink keeps its own call. A second IMEI does not wait on that call or close it. Replies for the first device are published to devices/490154203237518/downlink. {topic.N} is 1-based, so the device id is segment 2.

USE_TLS "false" is for a service that speaks plaintext HTTP/2. The default is TLS on port 443. Use 127.0.0.1 only when that service listens on the broker host. Set ADDRESS with GET ENV and credentials with GET SECRET.

How a Call Is Chosen

One connector is one channel and one RPC method (SERVICE + METHOD). HTTP/2 carries many calls on that channel. Add another connector when you need a second method.

LoT	What it sets
`ADDRESS`, `PORT`, `USE_TLS`	Channel to the service
`SERVICE`, `METHOD`	Which RPC to call
`RPC_TYPE`	Unary, server stream, client stream, or bidirectional
`SOURCE_TOPIC`	MQTT topic that supplies request bytes
`DESTINATION_TOPIC`	MQTT topic that receives response bytes
`SESSION_SCOPE`	Whether streaming calls are per topic or shared by the event
`CLIENT_AUTH`, `METADATA`	Initial metadata on the call, including `authorization`

There is no session-key setting. Put application ids inside the payload. The connector correlates calls with the MQTT topic match. ADD ROUTE_CONFIG still controls wait-or-drop when this connector falls behind incoming MQTT. See Store and Forward Configuration.

GRPC_CONFIG

Parameter	Required	Default	Description
`ADDRESS`	Yes	—	Host name or IP of the gRPC service. A `dns:///`, `https://`, or `http://` prefix is removed. A `:port` suffix overrides `PORT` (do not append `:port` to a bracketed IPv6 address — set `PORT` instead).
`PORT`	No	`443`	Channel port, unless `ADDRESS` already includes a port.
`USE_TLS`	No	`true`	`true` uses TLS. `false` uses plaintext HTTP/2. The scheme in `ADDRESS` does not override this.
`SERVICE`	Yes	—	Fully qualified service name (`package.Service`).
`METHOD`	Yes	—	RPC method name.
`RPC_TYPE`	No	`UNARY`	`UNARY`, `SERVER_STREAM`, `CLIENT_STREAM`, or `BIDI`.
`DEADLINE`	No	`30`	Call deadline in seconds, counted from when the call opens. Values below `1` are treated as 1 second. The deadline is not extended by later messages on the same call.
`IDLE_TIMEOUT`	No	`5`	Seconds of silence before the connector half-closes a client stream or bidirectional call. `0` disables that idle close.
`SESSION_SCOPE`	No	`TOPIC`	`TOPIC`: one streaming call per source topic. `EVENT`: one bidirectional call shared by every match of that event.
`METADATA`	No	—	Extra initial metadata. A JSON object, or `key=value,key2=value2`. Keys must not start with `grpc-`.
`CLIENT_AUTH`	No	`NONE`	Outbound authentication. See Authentication.

The parser also accepts SERVERSTREAM, CLIENTSTREAM, BIDIRECTIONAL, and DUPLEX for RPC_TYPE, and SOURCE, SOURCE_TOPIC, SHARED, and ROUTE for SESSION_SCOPE. Use the spellings in the table.

Event parameters

Parameter	Required	Description
`SOURCE_TOPIC`	Yes	MQTT topic (wildcards allowed) whose payload is the request bytes.
`DESTINATION_TOPIC`	Yes	MQTT topic for response bytes. `{topic.N}` expands from the source topic that triggered the call (1-based).

Several ADD EVENT blocks share the same channel and the same method. Each event keeps its own calls. Response publishes use QoS 0 and are not retained.

RPC Types

`RPC_TYPE`	MQTT to the service	Service to MQTT	How the client side ends
`UNARY`	One message, one call	One response	The call completes
`SERVER_STREAM`	One message, one call	Each response is published until the server finishes	The server finishes the call
`CLIENT_STREAM`	Many messages on the same call	One response after the client half-closes	Empty payload, or `IDLE_TIMEOUT`
`BIDI`	Many messages on the same call, while replies arrive	Each response is published as it arrives	Empty payload, or `IDLE_TIMEOUT`

An empty MQTT payload is a real empty request for UNARY and SERVER_STREAM. For CLIENT_STREAM and BIDI it is the half-close (no more client messages). The connector then waits up to 30 seconds for the remaining server bytes so the last response can still be published. PAYLOAD AS BYTES in an Action does not publish an empty payload. Send the zero-length message from an MQTT client (MQTT Explorer or any MQTT client) on that source topic. Raise DEADLINE above the longest call you expect. The default of 30 seconds ends a bidirectional session that runs longer than that, even when messages are still flowing. Raise IDLE_TIMEOUT when silence between messages is normal, or set it to 0 so only an empty payload half-closes the call.

Client stream

Publish request bytes to grpc/upload/in. Publish an empty payload, or wait for IDLE_TIMEOUT, to half-close. The single response is published to grpc/upload/out.

DEFINE ROUTE UploadGrpc WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '50051'
        WITH USE_TLS "false"
        WITH SERVICE "echo.Echo"
        WITH METHOD "ClientStream"
        WITH RPC_TYPE "CLIENT_STREAM"
        WITH IDLE_TIMEOUT '10'
    ADD EVENT Upload
        WITH SOURCE_TOPIC "grpc/upload/in"
        WITH DESTINATION_TOPIC "grpc/upload/out"

One shared call (`SESSION_SCOPE EVENT`)

SESSION_SCOPE "EVENT" keeps a single bidirectional call for every topic that matches the event. A new IMEI does not open its own call. Replies are published with {topic.N} taken from the latest source topic, not from the topic that opened the call.

DEFINE ROUTE SharedBidi WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '443'
        WITH USE_TLS "true"
        WITH SERVICE "echo.Echo"
        WITH METHOD "Bidi"
        WITH RPC_TYPE "BIDI"
        WITH DEADLINE '300'
        WITH IDLE_TIMEOUT '30'
        WITH SESSION_SCOPE "EVENT"
    ADD EVENT Converse
        WITH SOURCE_TOPIC "devices/+/uplink"
        WITH DESTINATION_TOPIC "devices/{topic.2}/downlink"

Leave the default TOPIC when each device must stay isolated. Use EVENT only when the service expects one shared stream.

After a bidirectional call has already ended, further payloads on that session are dropped until you publish an empty payload. The next non-empty payload then opens a fresh call. A service that ends the turn inside the protobuf, without an empty MQTT message, leaves that topic dropped until the empty payload arrives. The connector logs a warning with the topic and the number of bytes it dropped. That end-of-stream condition (InvalidArgument or Aborted) does not by itself mark the connector unhealthy.

Authentication

CLIENT_AUTH attaches credentials as initial metadata on each call. Access tokens are not published on MQTT topics. If the service returns unauthenticated, the connector invalidates the cached token, drops the streaming call, and retries that payload once so the new token is on the new call.

`CLIENT_AUTH`	What you set	How it is sent
`NONE`	—	No credential metadata. Default.
`API_KEY`	`API_KEY`. Optional `AUTH_HEADER` (default `X-API-Key`).	That metadata key
`BEARER`	`BEARER_TOKEN`	`authorization: Bearer …`
`OAUTH2_CLIENT_CREDENTIALS`	`CLIENT_ID`, `CLIENT_SECRET`, and `TOKEN_URL` or `ISSUER`. Optional `SCOPE`, `TOKEN_AUTH`.	`authorization: Bearer …` with a cached access token
`OAUTH2_REFRESH_TOKEN`	`REFRESH_TOKEN`, plus the same token endpoint fields	`authorization: Bearer …`
`OAUTH2_JWT_BEARER`	`SERVICE_ACCOUNT` (service-account JSON with `client_email` and `private_key`). Optional `TOKEN_URL` and `SCOPE`.	`authorization: Bearer …`
`GOOGLE_METADATA`	Optional `METADATA_URL` and `SCOPE`	`authorization: Bearer …` from the instance metadata server

TOKEN_AUTH chooses how the client authenticates to the token endpoint: CLIENT_SECRET_BASIC (default), CLIENT_SECRET_POST, or PRIVATE_KEY_JWT (CLIENT_ID plus CLIENT_ASSERTION_KEY as PEM or JWK). ISSUER is the OIDC issuer; the connector discovers the token endpoint when TOKEN_URL is omitted. BEARER_TOKEN and API_KEY also accept the setting name AUTH_SECRET. For OAUTH2_JWT_BEARER, set TOKEN_URL when the service-account JSON has no token_uri. For GOOGLE_METADATA, the default metadata address is http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token. Set METADATA_URL to point elsewhere. When SCOPE is set and that URL has no query string, the connector adds the scope as a scopes query parameter.

API key
OAuth client
Service account

Send a static key in initial metadata:

DEFINE ROUTE ApiKeyGrpc WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '443'
        WITH USE_TLS "true"
        WITH SERVICE "echo.Echo"
        WITH METHOD "Unary"
        WITH RPC_TYPE "UNARY"
        WITH CLIENT_AUTH "API_KEY"
        WITH API_KEY GET SECRET "GRPC_API_KEY"
        WITH AUTH_HEADER "X-API-Key"
    ADD EVENT CallEcho
        WITH SOURCE_TOPIC "grpc/echo/in"
        WITH DESTINATION_TOPIC "grpc/echo/out"

Fetch an access token with client credentials. Store the secret with -setSecret, not in the connector text:

DEFINE ROUTE OAuthGrpc WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '443'
        WITH USE_TLS "true"
        WITH SERVICE "acme.Device"
        WITH METHOD "Bidi"
        WITH RPC_TYPE "BIDI"
        WITH DEADLINE '300'
        WITH CLIENT_AUTH "OAUTH2_CLIENT_CREDENTIALS"
        WITH TOKEN_URL GET ENV "GRPC_TOKEN_URL"
        WITH CLIENT_ID GET ENV "GRPC_CLIENT_ID"
        WITH CLIENT_SECRET GET SECRET "GRPC_CLIENT_SECRET"
        WITH SCOPE "devices.stream"
    ADD EVENT Stream
        WITH SOURCE_TOPIC "devices/+/uplink"
        WITH DESTINATION_TOPIC "devices/{topic.2}/downlink"

DEFINE ROUTE JwtGrpc WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '443'
        WITH USE_TLS "true"
        WITH SERVICE "acme.Device"
        WITH METHOD "Bidi"
        WITH RPC_TYPE "BIDI"
        WITH DEADLINE '300'
        WITH CLIENT_AUTH "OAUTH2_JWT_BEARER"
        WITH TOKEN_URL GET ENV "GRPC_TOKEN_URL"
        WITH SERVICE_ACCOUNT GET SECRET "GRPC_SERVICE_ACCOUNT"
        WITH SCOPE "devices.stream"
    ADD EVENT Stream
        WITH SOURCE_TOPIC "devices/+/uplink"
        WITH DESTINATION_TOPIC "devices/{topic.2}/downlink"

Extra metadata is merged with the auth entry. Prefer a JSON object when a value contains a comma:

DEFINE ROUTE MetaGrpc WITH TYPE GRPC
    ADD GRPC_CONFIG
        WITH ADDRESS GET ENV "GRPC_HOST"
        WITH PORT '443'
        WITH USE_TLS "true"
        WITH SERVICE "echo.Echo"
        WITH METHOD "Unary"
        WITH RPC_TYPE "UNARY"
        WITH CLIENT_AUTH "BEARER"
        WITH BEARER_TOKEN GET SECRET "GRPC_BEARER"
        WITH METADATA "{\"x-tenant\":\"plant-a\"}"
    ADD EVENT CallEcho
        WITH SOURCE_TOPIC "grpc/echo/in"
        WITH DESTINATION_TOPIC "grpc/echo/out"

The connector fetches the access token when it starts. A token or channel failure leaves the connector disconnected. Startup waits up to 10 seconds for the channel. If the route has not started, the broker retries in the background, from 2 seconds up to 60 seconds, until the route starts or you remove it. Removing the connector drops the cached token. There is no setting for a client certificate or a private CA file. TLS uses the trust store on the broker host.

Payloads

The connector sends and publishes the MQTT payload as raw bytes. It does not encode JSON and it does not compile protobuf. A binary publish stays binary. If the topic cache has no raw bytes, the connector sends the payload text as UTF-8, which corrupts protobuf and audio. Publish protobuf with PUBLISH MODEL, or forward an existing blob with PAYLOAD AS BYTES. Decode replies with GET PROTO on the destination topic. The model describes the request. The Action writes binary protobuf to the topic the connector is listening on:

DEFINE MODEL EchoRequest WITH FORMAT PROTOBUF COLLAPSED
    ADD STRING "text" PROTO_TAG 1

DEFINE ACTION SendEcho
ON TOPIC "commands/echo" DO
    PUBLISH MODEL EchoRequest TO "grpc/echo/in" WITH
        text = (GET JSON "text" IN PAYLOAD AS STRING)

EchoGrpc from the Quick Start then forwards those bytes and publishes the service’s reply on grpc/echo/out. Read fields from that reply with GET PROTO and a response model. See Protobuf models.

Check the Round Trip

Deploy the connector

Deploy the LoT, then confirm the connector reaches the service. Health is retained on $SYS/Coreflux/Routes/EchoGrpc/status. See Connector status.

Subscribe to the reply

In MQTT Explorer, subscribe to the destination topic (grpc/echo/out, or devices/+/downlink for the per-device example).

Publish a request

Publish a payload to the source topic. For a unary echo, the reply is one message. For a bidirectional device topic, further publishes on that same topic reuse the call until an empty payload or the idle timeout.

Half-close a stream

For CLIENT_STREAM or BIDI, publish a zero-length payload on that source topic. The last server bytes are published, and the next non-empty payload opens a new call.

Best Practices

Keep SESSION_SCOPE "TOPIC" (the default) so each device topic, such as one IMEI, has its own call.
Set DEADLINE to the full length of a streaming call, not the gap between messages.
Store keys with GET SECRET. Do not publish tokens on MQTT topics.
Put correlation ids in the payload. The connector will not add a session id for you.
Tell operators that an empty payload half-closes CLIENT_STREAM and BIDI.
Do not send metadata keys that start with grpc-.

Limitations

Raw bytes only. The connector does not load a .proto or a protobuf codec. Encode before publish and decode after the reply.
One method per connector. SERVICE, METHOD, and RPC_TYPE are connector-wide.
Deadline is fixed when the call opens. A 30-second deadline stops a longer bidirectional call.
Finished bidirectional calls drop new bytes until an empty payload half-closes that session.
SESSION_SCOPE "EVENT" shares one call. Replies follow the latest source topic.
No client certificate or private CA field. TLS uses the broker host trust store.
Replies are QoS 0 and not retained.
Empty binary publishes from LoT do not half-close. PAYLOAD AS BYTES does not publish an empty payload. Publish the zero-length message from an MQTT client.

Troubleshooting

The connector will not start

GRPC_CONFIG needs ADDRESS, SERVICE, and METHOD. Each event needs SOURCE_TOPIC and DESTINATION_TOPIC.
The channel must connect within 10 seconds. Use a host and port the broker host can reach, and match USE_TLS to the service (plaintext often uses port 50051; TLS often uses 443).
If the route fails to start, the broker retries in the background, from 2 seconds up to 60 seconds, until the route starts or you remove it.
OAuth modes fetch a token at startup. Check TOKEN_URL or ISSUER, and the secret names.

Unauthenticated replies

The connector retries once after refreshing the token and opening a new streaming call. If that retry fails, check CLIENT_AUTH, the token endpoint, and that the secret resolves. Tokens are not written to MQTT topics.

Binary payloads look corrupted

Publish raw bytes (PUBLISH MODEL or PAYLOAD AS BYTES). A text payload is sent as UTF-8.

Metadata is rejected

Remove any key that starts with grpc-. Use a JSON object for METADATA when a value contains a comma.

The stream never finishes

Nothing published an empty payload, and IDLE_TIMEOUT is high or 0. Publish a zero-length payload on the source topic, or lower IDLE_TIMEOUT.

MQTT publishes continue but replies stop

The previous bidirectional call already ended. Payloads on that topic are dropped until an empty payload. Publish the empty payload, then send the next request. Check connector logs for the dropped-byte warning.

Replies land on the wrong device topic

SESSION_SCOPE "EVENT" expands {topic.N} from the latest source topic. Use TOPIC when each IMEI needs its own call and its own reply topic.

Next Steps

Protobuf Models

Encode request bytes with PUBLISH MODEL and read replies with GET PROTO.

Kafka Connector

Stream MQTT to an event log when the other side is Kafka, not a gRPC method.

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Call a gRPC Service from MQTT

When to use it

Quick Start

How a Call Is Chosen

GRPC_CONFIG

Event parameters

RPC Types

Client stream

One shared call (`SESSION_SCOPE EVENT`)

Authentication

Payloads

Check the Round Trip

Best Practices

Limitations

Troubleshooting

Next Steps

Protobuf Models

Kafka Connector

​Call a gRPC Service from MQTT

​When to use it

​Quick Start

​How a Call Is Chosen

​GRPC_CONFIG

​Event parameters

​RPC Types

​Client stream

​One shared call (SESSION_SCOPE EVENT)

​Authentication

​Payloads

​Check the Round Trip

​Best Practices

​Limitations

​Troubleshooting

​Next Steps

Protobuf Models

Kafka Connector

Call a gRPC Service from MQTT

When to use it

Quick Start

How a Call Is Chosen

GRPC_CONFIG

Event parameters

RPC Types

Client stream

One shared call (`SESSION_SCOPE EVENT`)

Authentication

Payloads

Check the Round Trip

Best Practices

Limitations

Troubleshooting

Next Steps