> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# gRPC Connector

> Call an external gRPC service from the Coreflux broker with TYPE GRPC — unary, streaming, and per-device bidirectional sessions

## 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.

> [!WARNING]
>
> 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](/latest/mqtt-broker/mcp-server). The hosted [documentation MCP](/ai/mcp) searches docs.coreflux.org and does not open a gRPC channel.

> [!TIP]
>
> **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.

> [!NOTE]
>
> 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](../rest-api-routes) for HTTP. Use the [Kafka connector](./kafka) for an event log. Encode and decode protobuf with [LoT models](/latest/lot-language/models/protobuf) — 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**

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

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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`.

**Server stream**

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

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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.

**One call per device**

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

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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.

> [!NOTE]
>
> `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`](/latest/mqtt-broker/secrets-and-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](../store-and-forward).

***

## 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](#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`.

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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.

```lot wrap focus={11} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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.

> [!WARNING]
>
> 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**

Send a static key in initial metadata:

```lot wrap focus={9-11} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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"
```

**OAuth client**

Fetch an access token with client credentials. Store the secret with [`-setSecret`](/latest/mqtt-broker/secrets-and-env), not in the connector text:

```lot wrap focus={10-14} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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"
```

**Service account**

Sign in with a service-account JSON key:

```lot wrap focus={10-13} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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:

```lot wrap focus={9-11} theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE MODEL EchoRequest WITH FORMAT PROTOBUF COLLAPSED
    ADD STRING "text" PROTO_TAG 1
```

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
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](/latest/lot-language/models/protobuf).

***

## Check the Round Trip

**1. 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](../overview#connector-status-monitoring).

**2. Subscribe to the reply**

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

**3. 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.

**4. 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

> [!WARNING]
>
> * **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](/latest/lot-language/models/protobuf)**
>
> Encode request bytes with `PUBLISH MODEL` and read replies with `GET PROTO`.

> **[Kafka Connector](./kafka)**
>
> Stream MQTT to an event log when the other side is Kafka, not a gRPC method.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.