connectrpc User Guide

This guide is the long-form companion to the crate README. It covers installation, code generation, server and client usage, streaming, tower middleware, TLS, error handling, and compression. If you just want to try the library, start with the README quick start and the examples/ directory.

Installation
Quick start
Code generation
Implementing servers
Streaming RPCs
Tower middleware
Interceptors
Hosting
Health checking
Server reflection
Production hardening
Clients
Errors and status codes
Compression
Examples directory tour

Installation

connectrpc ships as five crates:

Crate	Purpose
`connectrpc`	Tower-based runtime: server dispatcher, client transports, codec, compression
`protoc-gen-connect-rust` (binary, in `connectrpc-codegen`)	`protoc` plugin that generates service stubs
`connectrpc-build`	`build.rs` integration that runs the codegen at build time
`connectrpc-health`	The standard `grpc.health.v1.Health` service for liveness / readiness probes (Health checking)
`connectrpc-reflection`	The standard gRPC server reflection service (`grpc.reflection.v1` + `v1alpha`) for `grpcurl` / `buf curl` / Postman / `grpcui` (Server reflection)

Generated code references a small set of crates from your namespace, so a working Cargo.toml needs more than the runtime itself — this is the complete dependency block for a typical (JSON-capable) service:

[dependencies]
connectrpc = "0.9"
buffa = { version = "0.9", features = ["json"] }
buffa-types = { version = "0.9", features = ["json"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

[build-dependencies]
connectrpc-build = "0.9"

The buffa/serde entries come from buffa's generated message types. For proto-only builds (no JSON), drop the json features and serde/serde_json — see Generated Code Dependencies in the README.

MSRV

The MSRV is Rust 1.88, declared on the workspace and verified in CI. The crate uses Rust 2024 edition.

Feature flags

The runtime is feature-gated so you only pay for what you use:

Feature	Default	What it adds
`json`	yes	JSON codec for protobuf messages (the proto3-JSON wire format). Disabling it drops the `serde` requirement on message types — see Proto-only builds
`gzip`	yes	Gzip compression via `flate2`
`zstd`	yes	Zstandard compression via `zstd`
`client`	no	HTTP client transports (cleartext)
`client-tls`	no	TLS for client transports
`server`	no	Built-in hyper server (`Server`)
`server-tls`	no	TLS for the built-in server
`tls`	no	Convenience alias for both `server-tls` + `client-tls`
`axum`	no	Axum integration (`Router::into_axum_service`, `Router::into_axum_router`)

Common combinations:

# Just the server, behind axum
connectrpc = { version = "0.9", features = ["axum"] }

# Server + client, both with TLS
connectrpc = { version = "0.9", features = ["axum", "client", "tls"] }

# Built-in server (no axum)
connectrpc = { version = "0.9", features = ["server"] }

# Minimal (wasm-friendly: no networking, no native compression)
connectrpc = { version = "0.9", default-features = false }

Proto-only (no-JSON) builds

The Connect protocol supports two message codecs: binary proto and proto3 JSON. The JSON codec needs every message type to be serde::Serialize / Deserialize, which is why the code generator derives those impls by default. A deployment that only ever speaks binary proto can turn JSON off and shed those derives — smaller generated code, no serde_derive in the message-type build.

It takes two coordinated settings:

Generate without serde derives. Pass the no_json plugin option (or connectrpc-build's .generate_json(false)), so message structs are emitted without #[derive(serde::Serialize, serde::Deserialize)].

Disable the runtime json feature, which relaxes the message-type bounds from Message + Serialize/DeserializeOwned to just Message:

# Proto-only server: no JSON codec, no serde on message types.
# `default-features = false` is the only way to drop `json`, so it also drops
# the default compression features (`gzip`/`zstd`) — re-list the ones you
# still want.
connectrpc = { version = "0.9", default-features = false, features = ["server", "gzip", "zstd"] }

With json off, the Message + serde requirement is replaced by the JsonSerialize / JsonDeserialize marker traits, which become empty bounds — so a serde-free generated type still satisfies every handler, router, and client signature.

A proto-only server rejects JSON at content negotiation, before it touches the request body: application/json and application/connect+json (and the Connect GET encoding=json parameter) are unsupported media types, so the server responds with a bodyless HTTP 415 Unsupported Media Type (the client maps the status to an error code); application/grpc+json and application/grpc-web+json get a gRPC error status. Message-level encode/decode also returns Unimplemented as a defense-in-depth backstop. Handler-level errors (and the streaming end-of-stream frame) remain JSON, as the Connect spec requires regardless of the request codec. On the client side, the ClientConfig::json shorthand is removed from the API in a proto-only build, so JSON cannot be selected by mistake.

connectrpc itself still depends on serde and serde_json even in a proto-only build — the always-JSON error wire format needs them — so they stay in cargo tree. What proto-only mode removes is the serde derive on your generated message types and the per-message JSON (de)serialization paths.

Because json is an additive, default-on Cargo feature, it is only truly off when every crate in your dependency graph that depends on connectrpc disables it. If any other crate pulls in connectrpc with json enabled, feature unification turns it back on for the whole build, the markers revert to Serialize/DeserializeOwned, and your serde-free generated types stop compiling (Serialize is not satisfied — note the error names the trait, not the feature). Proto-only mode therefore fits a leaf binary or a fully proto-only graph, not one library inside a mixed workspace.

View-body responses are already proto-only and return Unimplemented for the JSON codec — see Returning a view body — so a proto-only build changes nothing for them.

Quick start

Define a service:

// proto/greet.proto
syntax = "proto3";
package greet.v1;

service GreetService {
  rpc Greet(GreetRequest) returns (GreetResponse);
}

message GreetRequest { string name = 1; }
message GreetResponse { string greeting = 1; }

Generate code with connectrpc-build in build.rs:

[build-dependencies]
connectrpc-build = "0.9"

// build.rs
fn main() {
    connectrpc_build::Config::new()
        .files(&["proto/greet.proto"])
        .includes(&["proto/"])
        .include_file("_connectrpc.rs")
        .compile()
        .unwrap();
}

Implement the service:

// src/main.rs
use std::sync::Arc;
use connectrpc::{RequestContext, Response, Router, ServiceRequest, ServiceResult};

pub mod proto {
    connectrpc::include_generated!();
}
use proto::greet::v1::*;

struct MyGreet;

impl GreetService for MyGreet {
    async fn greet(
        &self,
        _ctx: RequestContext,
        req: ServiceRequest<'_, GreetRequest>,
    ) -> ServiceResult<GreetResponse> {
        Response::ok(GreetResponse {
            greeting: format!("Hello, {}!", req.name),
            ..Default::default()
        })
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let router = Router::new().add_service(Arc::new(MyGreet));
    let app = router.into_axum_router();

    let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

That's the full server. Make a request with curl to confirm it works:

curl -X POST http://localhost:8080/greet.v1.GreetService/Greet \
  -H 'content-type: application/json' \
  -d '{"name": "World"}'

For runnable end-to-end examples, see the examples/ directory.

Code generation

Two workflows are supported. Both produce the same runtime API.

`connectrpc-build` (build-time, simplest)

Used in build.rs. Compiles .proto files at build time, regenerates on change, no extra binaries needed.

// build.rs
fn main() {
    connectrpc_build::Config::new()
        .files(&["proto/greet.proto", "proto/billing.proto"])
        .includes(&["proto/"])
        .include_file("_connectrpc.rs")
        .compile()
        .unwrap();
}

Output is unified: message types and service stubs in one file per proto, included into your crate with connectrpc::include_generated!(). Best for simple projects.

If you need the compiled FileDescriptorSet at runtime — most commonly to feed the connectrpc-reflection crate — chain .emit_descriptor_set("svc_descriptor.bin") before .compile(). The name must be a bare file name (no path separators). The set (including the full transitive import closure) is written to OUT_DIR and can be embedded with include_bytes!(concat!(env!("OUT_DIR"), "/svc_descriptor.bin")). See the Config::emit_descriptor_set rustdoc for details.

`buf generate` (checked-in code, production-grade)

Recommended when you want generated code committed to the repo, multi-output structure (e.g. separate proto modules from service modules), or when generating across language boundaries from one schema. Requires three plugins: protoc-gen-buffa for message types, protoc-gen-connect-rust for service stubs, and protoc-gen-buffa-packaging for assembling mod.rs trees.

protoc-gen-buffa owns <stem>.rs and its ancillary companion files (<stem>.__view.rs, <stem>.__oneof.rs, …); protoc-gen-connect-rust adds <stem>.__connect.rs containing the service trait + client. Each package gets a <pkg>.mod.rs stitcher that include!s all of them.

If you'd rather have one file per proto package — the convention that Buf Schema Registry cargo SDK generation and tonic-style build integrations expect — pass opt: file_per_package to both protoc-gen-buffa and protoc-gen-connect-rust. That collapses each plugin's output to one <dotted.pkg>.rs per package with everything inlined and no per-file companion files or <pkg>.mod.rs stitcher. Drop the protoc-gen-buffa-packaging invocations under this layout — there is nothing for them to wire — and either let your downstream tool synthesise the module tree from <dotted.package>.rs filenames (BSR cargo SDKs do this automatically) or hand-write the mod.rs. Keep routing each plugin to its own out: directory; the filename is shared between them and would silently overwrite in a shared one. connectrpc-build users get the same option as Config::file_per_package(true), which inlines the service stubs into buffa's <dotted.pkg>.rs and is otherwise transparent — the include file picks up the new filename automatically.

See the README's Code generation section for plugin installation, buf.gen.yaml configuration, and the buffa_module shorthand for cross-tree references.

Inclusion patterns side-by-side

Both workflows produce the same runtime API. The only difference is how you include the generated code into your crate:

// connectrpc-build (build.rs) users:
pub mod proto { connectrpc::include_generated!(); }

// buf generate users:
#[path = "generated/proto/mod.rs"]
pub mod proto;

The underlying difference (OUT_DIR vs a known source path) is honest and visible, but the call-site shape is parallel.

Very large schemas

Generation reads a descriptor set, and buffa bounds how much memory a decode may commit to repeated elements. That bound is an amplification defence sized for untrusted wire input, and it is charged on each element's struct size rather than on its encoded bytes — so descriptor sets, whose structs are wide, reach it while still looking small on the wire. The tooling paths — this plugin and connectrpc-build — therefore decode under buffa's much higher tooling bound of 1 GiB. It stays finite, so a truncated or corrupt set still fails with an error instead of exhausting memory.

If you do exceed it, generation stops with a message naming the budget in force and both ways to raise it. Both accept a byte count or unlimited, and they differ in reach.

The environment variable covers the whole run, which is normally what you want:

BUFFA_ELEMENT_MEMORY_LIMIT=4294967296 buf generate
BUFFA_ELEMENT_MEMORY_LIMIT=4294967296 cargo build

buf generate hands the identical request to every plugin in the run, so a schema big enough to need raising needs it for protoc-gen-buffa and protoc-gen-connect-rust alike. The variable is buffa's rather than a connect-specific twin precisely so that one setting serves both. It is also the only override that reaches connectrpc-build, since a build script has no plugin parameter string.

The plugin option is per-plugin. buf gives each plugin its own opt list, so this raises the bound for protoc-gen-connect-rust and nothing else — every other plugin in the run needs its own entry:

plugins:
  - local: protoc-gen-buffa
    out: src/generated
    opt:
      - element_memory_limit=4294967296
  - local: protoc-gen-connect-rust
    out: src/generated
    opt:
      - element_memory_limit=4294967296

Prefer a byte count to unlimited. unlimited removes the ceiling entirely, so a truncated or corrupt descriptor set stops failing with an error and starts exhausting memory instead — on CI that reads as a flaky runner rather than a bad input. A generous number keeps the diagnostic.

This bound governs generation only, and nothing here reaches a running server. A descriptor set handed to a Reflector decodes on buffa's untrusted-input default with no override: reflection descriptors can come from a peer rather than your own build, so the defence stays on. A set over that budget reports ReflectionError::ElementBudget, and the remedy is a smaller set — strip source_code_info, or narrow it to the files that server reflects.

Request decoding is a different budget again, configured per service through Limits::element_memory_limit. None of the three read each other, so raising the build-time bound has no effect on either.

Implementing servers

A service is a Rust trait generated from your .proto file. The trait name is the proto service name in UpperCamelCase (GreetService becomes trait GreetService; greet_service would too), and each RPC becomes an async method.

Every service in a proto package is generated into one Rust module, so the names derived from them must be distinct after that normalization: two services that differ only in case or underscores (XGet and X_Get), or a service and method pair whose words split differently from another's (XGet.Foo and X.GetFoo both name the X_GET_FOO_SPEC constant), are rejected at generation time with a message naming both sides. Rename one of them in the proto.

Handler signatures

Unary handlers take a read-only RequestContext plus a borrowed ServiceRequest<'_, RequestType>, and return ServiceResult<ResponseType>:

impl GreetService for MyGreet {
    async fn greet(
        &self,
        _ctx: RequestContext,
        req: ServiceRequest<'_, GreetRequest>,
    ) -> ServiceResult<GreetResponse> {
        // req derefs to the request view: zero-copy field access.
        // String fields are &str borrowed from the request buffer.
        Response::ok(GreetResponse {
            greeting: format!("Hello, {}!", req.name),
            ..Default::default()
        })
    }
}

The ServiceRequest shape lets handlers read string fields without allocating - req.name is a &str directly into the request bytes, and the borrow may be held across .await points. The request is borrowed from the dispatcher-owned body, so the response (and anything moved into tokio::spawn) cannot borrow from it - call .to_owned_message() to get the owned struct when you need one. The conversion is infallible: buffa charges every unknown-field record against the decode-time allowance, so a request that decoded successfully always re-materializes.

`RequestContext` and `Response`

Request-side metadata lives on RequestContext (passed in); response-side metadata lives on Response<B> (returned):

RequestContext is #[non_exhaustive]; read it through the accessor methods (new request-scoped metadata can then be added in minor releases):

`RequestContext` accessor	Purpose
`ctx.header(name)` / `ctx.headers()`	Caller-supplied headers (after protocol-prefix stripping)
`ctx.deadline()`	Absolute `Instant` if the caller set a timeout
`ctx.time_remaining()`	Saturating `Option<Duration>` until the deadline (`None` when no deadline is set) — budget downstream calls with this
`ctx.extensions()`	`http::Extensions` carried from the underlying `http::Request`
`ctx.path()`	Requested procedure path (`/package.Service/Method`) from the request URI
`ctx.spec()`	Static metadata for the dispatched RPC method (`Spec`); `None` only for low-level manual registrations that do not attach one
`ctx.protocol()`	The negotiated wire protocol for this request (`Connect` / `Grpc` / `GrpcWeb`)
`ctx.peer_addr()`	Remote socket address (requires the `server` feature; `None` when the transport didn't insert it)
`ctx.peer_certs()`	TLS client cert chain (requires the `server-tls` feature; `None` for plaintext or no client cert)

For example, propagating the caller's deadline to a downstream RPC and reading the peer cert chain:

// Budget downstream calls from the remaining time, leaving a margin
// for response encoding and network round-trips.
if let Some(remaining) = ctx.time_remaining() {
    let budget = remaining.saturating_sub(Duration::from_millis(50));
    options = options.with_timeout(budget);
}

// Typed peer lookup — returns None instead of panicking when the
// request didn't arrive over mTLS.
if let Some(certs) = ctx.peer_certs() {
    authorize(certs)?;
}

`Response<B>` field	Purpose
`body`	The response message (or `ServiceStream<M>` for streaming)
`headers`	Headers to send before the body
`trailers`	Trailers to send after the body
`compress`	Override the server's compression policy for this RPC

ServiceResult<B> is Result<Response<B>, ConnectError>. The happy path is Response::ok(body); to attach response metadata, use the builder:

async fn greet(
    &self,
    _ctx: RequestContext,
    req: ServiceRequest<'_, GreetRequest>,
) -> ServiceResult<GreetResponse> {
    Ok(Response::new(GreetResponse { /* ... */ })
        .with_header("x-greet-version", "v2")
        .with_trailer("x-server-id", "node-7"))
}

RequestContext::extensions() is the passthrough channel for tower-layer state: a custom auth layer can stamp a UserId into the request's http::Extensions, and the dispatcher forwards that map verbatim into the request context for the handler to read with ctx.extensions().get::<UserId>(). For the well-known peer types, prefer the typed ctx.peer_addr() / ctx.peer_certs() accessors — they return None rather than panicking when the transport didn't insert them. See Tower middleware for the full pattern, and with_connection_extensions under TLS for state computed once per connection (TLS or plaintext) rather than per request.

What you see vs. what you write

The generated trait declares unary methods with the full RPITIT bounds:

fn say(&self, ctx: RequestContext, req: ...)
    -> impl Future<Output = ServiceResult<impl Encodable<SayResponse> + Send + 'static + use<Self>>> + Send;

That is what cargo doc and rust-analyzer hover show. You never write that form in an impl - async fn desugars the outer impl Future, and returning ServiceResult<SayResponse> (the concrete owned type) refines the impl Encodable<...> bound. The short form in the examples above is all you need.

The `refining_impl_trait` lint

The generated trait declares unary/client-stream returns as ServiceResult<impl Encodable<M>> so handlers can return either the owned M or a borrowed view that encodes as M (see below). Writing your impl as -> ServiceResult<FooResponse> refines that opaque bound to a concrete type, which triggers refining_impl_trait_internal / refining_impl_trait_reachable. This is intentional - the refinement is the point. Add at your crate root:

#![allow(refining_impl_trait_internal, refining_impl_trait_reachable)]

or #[allow(refining_impl_trait)] on the impl block.

Returning a view body

For handlers that often return the request unchanged (proxies, filters, validators), the Encodable<M> bound lets you skip the owned-message allocation by returning an OwnedView rebuilt zero-copy from the retained request bytes (the request itself is borrowed and cannot outlive the call, so it is re-decoded - a Bytes refcount bump plus a decode walk, with no per-field copy). Codegen emits OwnedFooView aliases and impl Encodable<Foo> for OwnedFooView per RPC type (or, with encodable_impls=all_messages, the impls for every message in the generated crate - the aliases stay RPC-scoped). (When two RPC types in the same package would alias to the same OwnedFooView name — e.g. a local MyMessage plus an imported api.v1.foo.bar.MyMessage — the alias is suppressed for both; spell the inlined OwnedView<…View<'static>> form for those types.) connectrpc::MaybeBorrowed covers the conditional case:

use connectrpc::{MaybeBorrowed, RequestContext, Response, ServiceRequest, ServiceResult};
// `Record` and `OwnedRecordView` come from your generated module.

async fn redact(
    &self,
    _ctx: RequestContext,
    req: ServiceRequest<'_, Record>,
) -> ServiceResult<MaybeBorrowed<Record, OwnedRecordView>> {
    if req.email.is_empty() && req.ssn.is_empty() {
        // Pass-through. The response must be 'static, so rebuild an
        // OwnedView from the retained body bytes - zero-copy (Bytes
        // refcount + decode walk), then re-encode via ViewEncode.
        return Response::ok(MaybeBorrowed::Borrowed(req.to_owned_view()));
    }
    let mut owned = req.to_owned_message();
    owned.email.clear();
    owned.ssn.clear();
    Response::ok(MaybeBorrowed::Owned(owned))
}

A view body's large bytes and string fields reach the transport by reference count rather than being copied into one buffer, on unary and streaming responses alike (each stream item is encoded the same way). Two things restore the copy: an owned-message body, whose fields the encoder cannot borrow, and compression, which needs one contiguous input. Under the default CompressionPolicy any response over 1 KiB to a client advertising gzip is compressed, so a handler streaming large, poorly-compressible view items should return Response::stream(..).compress(false) to keep the copy-free path.

The 'a on the trait method also lets the body borrow from &self (e.g. cached server state). View bodies only encode for the proto codec - JSON clients receive unimplemented; see MaybeBorrowed's codec note. View-body impls are not emitted for output types mapped via extern_path (the impl would be an orphan in the consuming crate) - the impls must live in the crate that owns the type. If you generate that crate yourself, regenerate it with encodable_impls=all_messages (see the protoc-gen-connect-rust option docs) and views of its types become returnable from any crate. For types you don't generate (e.g. well-known types from buffa-types), return the owned message or use PreEncoded::from_view.

Returning errors

Handlers return ConnectError for failures. Each error carries an ErrorCode (the canonical Connect/gRPC status), a message, optional structured details, and optional metadata (headers + trailers):

use connectrpc::{ConnectError, ErrorCode};

return Err(ConnectError::new(
    ErrorCode::NotFound,
    format!("user {name:?} not found"),
));

The dispatcher maps ErrorCode to the appropriate HTTP status and serializes the error in the protocol the caller is using (Connect JSON, Connect binary, gRPC trailers, or gRPC-Web). Handlers don't need to know which protocol the caller chose.

Registering services on a Router

let router = Router::new()
    .add_service(Arc::new(MyGreet))
    .add_service(Arc::new(MyBilling));

The generated register extension method remains available when the inside-out form is useful:

let router = Arc::new(MyGreet).register(Router::new());

To combine routers that were built separately, use Router::merge (owned, chainable), Router::merge_in_place (in place), or the merge_routers free function for many at once. Merging two routers that register the same method path panics by default, so an accidental collision fails loudly at startup; call Router::allow_overrides() first when last-wins replacement is intended:

let router = defaults.allow_overrides().merge(overrides);

When the routers come from dynamic input (a plugin list, config-driven service set) and a collision should be handled rather than crash the process, use Router::try_merge / Router::try_merge_in_place, which return a RouterMergeError listing the conflicting paths instead of panicking.

The router is what you mount on axum (router.into_axum_router()) or pass to the built-in Server.

Testing handlers

Handlers are plain async methods, so unit tests call them directly - no server, no sockets. Construct the inputs the same way the dispatcher does:

use buffa::Message;               // encode_to_vec / decode_from_slice
use buffa::view::HasMessageView;  // GreetRequest::decode_view

#[tokio::test]
async fn greet_uses_the_name() {
    let svc = GreetServiceImpl::default();

    // Unary: encode the request, decode a view over it, wrap the pair.
    let body = Bytes::from(GreetRequest {
        name: "ada".into(),
        ..Default::default()
    }.encode_to_vec());
    let view = GreetRequest::decode_view(&body).unwrap();
    let req = ServiceRequest::<GreetRequest>::from_parts(&view, &body);

    let resp = svc.greet(RequestContext::new(HeaderMap::new()), req)
        .await
        .unwrap();

    // The trait's response body is an opaque `impl Encodable<GreetResponse>`;
    // encode it (exactly what the dispatcher does) and decode to assert on
    // fields. Headers and trailers are directly accessible on `resp`. The
    // UFCS call avoids ambiguity with `buffa::Message::encode`, which is
    // also in scope.
    use connectrpc::Encodable;
    let bytes = Encodable::encode(&resp.body, CodecFormat::Proto).unwrap();
    let reply = GreetResponse::decode_from_slice(&bytes).unwrap();
    assert_eq!(reply.greeting, "Hello, ada!");
}

Streaming inputs are one call each: StreamMessage::from_message(&msg) builds an item, and futures::stream::iter([...]) boxed into an InboundStream builds the request stream:

let items = [Ok(StreamMessage::from_message(&SumRequest {
    value: Some(3), ..Default::default()
}))];
let requests: InboundStream<SumRequest> =
    Box::pin(futures::stream::iter(items));
let resp = svc.sum(RequestContext::new(HeaderMap::new()), requests).await?;

RequestContext::new takes the request headers; its with_* builders cover peer identity and other per-call inputs.

Streaming RPCs

ConnectRPC supports all four RPC types. Define them in your .proto file with the standard stream keyword:

service NumberService {
  rpc Square(SquareRequest) returns (SquareResponse);                 // unary
  rpc Range(RangeRequest) returns (stream RangeResponse);             // server stream
  rpc Sum(stream SumRequest) returns (SumResponse);                   // client stream
  rpc RunningSum(stream RunningSumRequest) returns (stream RunningSumResponse);  // bidi
}

The runnable demo for each type lives in examples/streaming-tour/. The handler signatures are summarized below.

The streaming-handler trait signatures use Pin<Box<dyn Stream<...> + Send>> for both inbound and outbound streams. That's verbose, so the snippets here use connectrpc::ServiceStream<T> (a boxed Send stream of Result<T, ConnectError>).

Server streaming

The handler returns a stream of responses. Use any futures::Stream you like, then wrap it with Response::stream_ok (or Ok(Response::stream(s).with_header(...)) if you need response metadata):

async fn range(
    &self,
    _ctx: RequestContext,
    req: ServiceRequest<'_, RangeRequest>,
) -> ServiceResult<ServiceStream<RangeResponse>> {
    let stream = futures::stream::iter(/* ... */);
    Response::stream_ok(stream)
}

Client streaming

The handler receives an InboundStream<Req> — a ServiceStream of StreamMessage<Req> items — and returns a single response. Each item owns its decoded buffer, is Send + 'static (so it can be buffered or moved into spawned tasks), and exposes zero-copy accessor methods per field:

async fn sum(
    &self,
    _ctx: RequestContext,
    mut requests: InboundStream<SumRequest>,
) -> ServiceResult<SumResponse> {
    let mut total: i64 = 0;
    while let Some(req) = requests.next().await {
        total += req?.value().unwrap_or(0) as i64;
    }
    Response::ok(SumResponse { total: Some(total), ..Default::default() })
}

The request stream yields Err(ConnectError) if the upload fails partway — a truncated body or broken transport — so a partial stream is not mistaken for a complete one. The req? in the loop above propagates that error as the RPC's failure, which is the right default for handlers that aggregate inbound messages. Only a clean None means the client finished the stream.

Bidirectional streaming

Takes a request stream and returns a response stream. Both sides can emit messages independently:

async fn running_sum(
    &self,
    _ctx: RequestContext,
    requests: InboundStream<RunningSumRequest>,
) -> ServiceResult<ServiceStream<RunningSumResponse>> {
    // Map the request stream to a response stream however you like.
    let response_stream = futures::stream::unfold(/* ... */);
    Response::stream_ok(response_stream)
}

For bidirectional streams that need true full-duplex behavior (server emits messages independently of client send rate), use a tokio::sync::mpsc channel: spawn a task that reads from requests and writes to the channel sender, return a ReceiverStream as the response. See tests/streaming/src/lib.rs for an example.

Calling streaming RPCs from a client

Generated clients expose a method for each RPC. Server streaming returns a stream you call .message().await? on; bidi returns a handle with .send(req).await? and .message().await? plus .close_send():

// Server streaming. Each item is a `StreamMessage` - the same wrapper
// server handlers receive for inbound streams. Read fields zero-copy
// via `.view()` (or the generated accessor methods), and convert with
// `.to_owned_message()` when you need the owned struct.
let mut stream = client.range(req).await?;
while let Some(msg) = stream.message().await? {
    println!("{}", msg.view().value.unwrap_or_default());
}

// Client streaming - takes an async `Stream` of requests, so messages
// can be produced as they become available without buffering the whole
// upload. A ready collection is adapted with `stream_iter` (a re-export
// of `futures::stream::iter`, so no direct `futures` dependency needed):
let resp = client
    .sum(connectrpc::stream_iter(vec![req1, req2, req3]))
    .await?;

// ...or feed the call from a live producer through a channel-backed
// stream (add the `tokio-stream` crate for the wrapper). The generated
// bound, `ClientRequestStream<T>`, is `Stream<Item = T> + Send + 'static`:
// the stream backs the request body, so yield owned messages rather than
// borrows of local data.
let (tx, rx) = tokio::sync::mpsc::channel(16);
tokio::spawn(async move {
    while let Some(chunk) = source.recv().await {
        if tx.send(request_for(chunk)).await.is_err() {
            break; // call ended — stop producing
        }
    }
});
let resp = client
    .sum(tokio_stream::wrappers::ReceiverStream::new(rx))
    .await?;

// Bidi - received items are `StreamMessage`s too
let mut bidi = client.running_sum().await?;
bidi.send(req).await?;
if let Some(reply) = bidi.message().await? {
    println!("{}", reply.view().total.unwrap_or_default());
}
bidi.close_send();

// For true full duplex, split the bidi stream into independently owned
// halves and drive them from separate tasks. Response-dependent sends
// require an HTTP/2 transport (on HTTP/1.1 no response arrives until the
// upload completes). Dropping the send half ends the upload cleanly;
// dropping the receive half cancels the RPC.
let (mut send, mut recv) = client.running_sum().await?.into_split();
let reader = tokio::spawn(async move {
    while let Some(reply) = recv.message().await? {
        println!("{}", reply.view().total.unwrap_or_default());
    }
    Ok::<_, connectrpc::ConnectError>(())
});
for req in requests {
    send.send(req).await?;
}
send.close_send();
reader.await.expect("reader task")?;

? on message() is the complete error handling: Ok(None) means the server finished cleanly, and a terminal RPC error — including a gRPC/gRPC-Web stream that ends without a usable grpc-status — comes back as Err, sticky across calls. The error() and trailers() accessors remain available afterwards for post-hoc inspection.

Dropping a client-streaming call cancels it: the request body is dropped with the future, so messages the stream had not yet yielded never reach the server. Wrapping such a call in a timeout therefore abandons the upload rather than truncating it cleanly — drive the call to completion whenever the request must be delivered.

Both streaming-tour/src/client.rs and the eliza example show these patterns end-to-end.

Tower middleware

The connect router is a tower::Service, so any tower layer composes on top. The full reference is in examples/middleware/, which uses an axum::middleware::from_fn for bearer-token auth and chains it with tower-http's TraceLayer and TimeoutLayer.

Composing layers

Use tower::ServiceBuilder for clear top-to-bottom ordering, mounted on axum::Router::layer() so axum handles the body conversion from ConnectRpcBody to axum::body::Body:

use std::sync::Arc;
use std::time::Duration;
use tower::ServiceBuilder;
use tower_http::{trace::TraceLayer, timeout::TimeoutLayer};

let connect_router = Router::new().add_service(service);
let tokens = Arc::new(token_table());
let app = axum::Router::new()
    .fallback_service(connect_router.into_axum_service())
    .layer(
        ServiceBuilder::new()
            .layer(TraceLayer::new_for_http())                  // outermost
            .layer(axum::middleware::from_fn_with_state(tokens, auth_middleware))
            .layer(TimeoutLayer::with_status_code(              // innermost
                http::StatusCode::REQUEST_TIMEOUT,
                Duration::from_secs(5),
            )),
    );

ServiceBuilder applies layers top-to-bottom: the first .layer() sees requests first (and responses last). A request flows trace -> auth -> timeout -> dispatcher -> handler.

For auth and similar interceptors, axum::middleware::from_fn (or from_fn_with_state for stateful cases) is usually the lightest path because it lets you write the middleware as a plain async function. A hand-rolled tower::Layer + tower::Service pair is also fine when you need finer control - both produce a Layer that ServiceBuilder accepts.

Passing data from a layer to a handler

The dispatch path moves the request's http::Extensions into the request context verbatim. So a middleware that inserts a value via req.extensions_mut().insert(value) makes that value available to the handler via ctx.extensions().get::<T>(). This is the canonical way to pass per-request state from middleware (auth identity, trace IDs, remote addr, TLS peer info) into the handler.

The middleware example does exactly this with a UserId:

// In the auth middleware:
req.extensions_mut().insert(UserId(user.into()));
next.run(req).await

// In the handler:
let user = ctx.extensions().get::<UserId>().unwrap();

Static method metadata (`Spec`)

Handlers and middleware can read which RPC method is being invoked without re-parsing the request URL. ctx.spec() returns an Option<Spec> describing the dispatched method: its fully-qualified procedure path, message-flow shape, the proto-declared idempotency contract, and whether the spec came from a server-side dispatcher or a generated client.

async fn greet(
    &self,
    ctx: RequestContext,
    req: ServiceRequest<'_, GreetRequest>,
) -> ServiceResult<GreetResponse> {
    if let Some(spec) = ctx.spec() {
        tracing::info_span!(
            "rpc",
            "rpc.system" = "connect_rpc",
            "rpc.service" = spec.service(),
            "rpc.method" = spec.method(),
        );
    }
    // ...
}

ctx.protocol() is the per-request companion: it returns the negotiated wire protocol (Connect, Grpc, or GrpcWeb) so an observability layer can label spans with rpc.system correctly. Spec carries only registration-time facts that are the same for every request to that method; per-request state lives on RequestContext. This mirrors connect-go's Spec / Peer split.

Spec is Copy, contains only 'static data, and is #[non_exhaustive] — destructure with a trailing ..:

use connectrpc::{Spec, SpecOrigin, StreamType, IdempotencyLevel};

let Spec { procedure, stream_type, origin, idempotency_level, .. } = spec;

Code generation also emits a pub const <SERVICE>_<METHOD>_SPEC: Spec per method that you can reference directly without a request in flight — useful for building static lookup tables, validating routing, or testing:

use crate::connect::greet::v1::GREET_SERVICE_GREET_SPEC;

assert_eq!(GREET_SERVICE_GREET_SPEC.procedure, "/greet.v1.GreetService/Greet");
assert_eq!(GREET_SERVICE_GREET_SPEC.stream_type, StreamType::Unary);
assert_eq!(GREET_SERVICE_GREET_SPEC.origin, SpecOrigin::Server);

The generated client passes the same constant to the runtime with its origin flipped, GREET_SERVICE_GREET_SPEC.with_origin(SpecOrigin::Client), so that is the value a client-side interceptor observes. Because Spec's PartialEq covers every field, that value is not == to the constant; an interceptor that runs on both sides and asks "is this the Greet method?" should use spec.same_method(GREET_SERVICE_GREET_SPEC) (or compare procedure), which ignores origin. The low-level connectrpc::client::call_* entry points take a Spec for the same reason; see the call_unary and Spec::client rustdoc for hand-written and dynamic callers.

Both dispatch paths populate ctx.spec(). A code-generated FooServiceServer<T> always supplies a Spec. The dynamic Router (used by FooServiceExt::register(Router)) does too — the generated register() chains .with_spec(SPEC_CONST) after each route. The only handlers that see ctx.spec() == None are those registered through low-level manual registration without attaching a Spec. ctx.path() is populated unconditionally regardless of dispatch path — use it when you only need the procedure name and want to be robust to a missing Spec.

Short-circuit responses

A layer can short-circuit by returning a response without invoking the inner service. The middleware example does this for unauthorized requests, returning a 401 with a Connect-protocol JSON error body so clients see the failure on the same code path they use for handler errors.

Interceptors

Tower middleware (above) operates on http::Request / http::Response — it's the right level for cross-cutting concerns that don't need to know they're wrapping an RPC: connection-scoped tracing, gzip, raw header manipulation. Interceptors are the typed RPC layer on top: async hooks per call. intercept_head runs before the request body is read; intercept_unary and intercept_streaming run after the request head is parsed and the body has been read and decompressed (under the service's Limits), but before the message is decoded, and before the handler. Interceptors see the resolved Spec, the parsed headers, the deadline, the negotiated protocol, the request extensions, and a lazily decoded message body — what a span builder, validator, rate limiter, or authorization check wants. A check that needs only the resolved Spec, the headers, and the connection's extensions can run earlier, in Interceptor::intercept_head, before any of the body is read (see Rejecting before the body is read). For authentication — rejecting a caller that has presented no credential — read what an unauthenticated request costs before choosing between the two.

use connectrpc::interceptor::{UnaryRequest, UnaryResponse};
use connectrpc::{ConnectError, Interceptor, Next, Payload, Response};

struct Logging;

#[connectrpc::async_trait]
impl Interceptor for Logging {
    async fn intercept_unary(
        &self,
        req: UnaryRequest,
        next: Next<'_>,
    ) -> Result<UnaryResponse, ConnectError> {
        let path = req.ctx.path().unwrap_or("<unknown>").to_owned();
        let started = std::time::Instant::now();
        let resp = next.run(req).await;
        tracing::info!(rpc = %path, elapsed = ?started.elapsed(), ok = resp.is_ok());
        resp
    }
}

let server = GreetServiceServer::new(GreetServiceImpl);
let service = ConnectRpcService::new(server).with_interceptor(Logging);

Annotate impls with the re-exported #[connectrpc::async_trait] — there is no separate async-trait dependency for downstream crates. The default impls are passthroughs, so you only override the hook you need. For one-off interceptors, the unary_interceptor and streaming_interceptor closure helpers skip the struct boilerplate.

Ordering and registration

with_interceptor registers in outermost-first order, matching connect-go's WithInterceptors: the first interceptor registered sees the request first and the response last.

.with_interceptor(A).with_interceptor(B)

request:   A → B → handler
response:  A ← B ← handler

A service with no interceptors registered pays one is_empty() branch on the dispatch path — no per-request allocation, no Payload construction, no Boxing.

To share one interceptor instance across several ConnectRpcServices (an authorization interceptor whose policy cache or rate-limit counter is process-wide), use with_interceptor_arc(Arc<dyn Interceptor>). with_interceptor allocates a fresh Arc per registration; with_interceptor_arc accepts the one you already hold.

Reading and rewriting the request

UnaryRequest is { ctx: RequestContext, payload: Payload }. Mutating ctx through ctx.headers_mut() or ctx.extensions_mut() before next.run propagates to the handler. The payload is the request body — wire bytes plus a lazy decode cache. Most interceptors never read it; ones that do call payload.message::<M>() to decode once and cache, so the handler's decode is free:

async fn intercept_unary(
    &self,
    mut req: UnaryRequest,
    next: Next<'_>,
) -> Result<UnaryResponse, ConnectError> {
    // Decode once; the handler reuses this decode via the Payload cache.
    let body = req.payload.message::<GreetRequest>()?;
    if body.name.is_empty() {
        return Err(ConnectError::invalid_argument("name is required"));
    }
    // Replace the body — the handler sees the replacement.
    let mut rewritten = body.clone();
    rewritten.name = rewritten.name.trim().to_owned();
    req.payload.set_message(rewritten);
    next.run(req).await
}

Short-circuiting and re-running

Returning without calling next.run() short-circuits the chain — neither inner interceptors nor the handler run. Returning Err surfaces the error on the protocol's normal error path, including any response_headers the error carries:

async fn intercept_unary(
    &self,
    req: UnaryRequest,
    next: Next<'_>,
) -> Result<UnaryResponse, ConnectError> {
    // Authentication already happened in Tower middleware, which stamped
    // the caller's identity into the request extensions before any body
    // byte was read. This interceptor decides whether that identity may
    // call this method.
    let user = req.ctx.extensions().get::<UserId>().cloned();
    let procedure = req.ctx.spec().map_or("", |s| s.procedure);
    if !self.policy.allows(user.as_ref(), procedure) {
        let mut err = ConnectError::permission_denied("not allowed");
        err.response_headers_mut().insert(
            "x-denied-by",
            http::HeaderValue::from_static("policy"),
        );
        return Err(err);
    }
    next.run(req).await
}

Returning Ok without calling next works too. Build the body from a typed message with Payload::from_message, which encodes lazily in the request's wire format:

if let Some(reply) = self.cache.get(req.payload.message::<LookupRequest>()?) {
    return Ok(Response::new(Payload::from_message(reply.clone(), req.payload.format())));
}
next.run(req).await

The opposite of short-circuiting is running the chain more than once. Next is Clone, and UnaryRequest::try_clone() copies the context and body, so a retry looks like this (gate it on Spec::idempotency_level; on the server a re-run invokes the handler again):

let spare = req.try_clone()?;
match next.clone().run(req).await {
    Err(e) if e.code == ErrorCode::Unavailable => next.run(spare).await,
    done => done,
}

Streaming RPCs

intercept_streaming covers server-streaming, client-streaming, and bidi with one Stream-shaped hook. It runs once at stream establishment — before any messages flow — and receives an inbound PayloadStream plus a NextStream<'_> continuation. The returned StreamResponse carries the outbound PayloadStream and response metadata.

use connectrpc::interceptor::{StreamRequest, StreamResponse};
use connectrpc::{Interceptor, NextStream, PayloadStream};

#[connectrpc::async_trait]
impl Interceptor for AuthzInterceptor {
    async fn intercept_streaming(
        &self,
        req: StreamRequest,
        inbound: PayloadStream,
        next: NextStream<'_>,
    ) -> Result<StreamResponse, ConnectError> {
        // Authorization runs once at establishment, not per message.
        self.check(&req.ctx)?;
        let resp = next.run(req, inbound).await?;
        Ok(resp.with_header("x-served-by", &self.node_id))
    }
}

To observe or transform individual messages, wrap inbound (or the returned resp.body) with a futures::Stream adapter — .map(), .then(), .filter(). There is no per-message send() call site to hook because Rust handlers return a Stream, they don't push into a connection. This is the same shape tower, tonic, and axum use for body interception. Cross-stream coordination (deciding on an outbound item based on what was observed inbound) needs shared state captured by both adapter closures (Arc<Mutex<..>>); this is rare — most interceptors observe one direction or none.

For server-streaming the inbound stream yields exactly one item; for client-streaming the outbound stream yields exactly one item. Read req.ctx.spec().map(|s| s.stream_type) to branch on cardinality.

Interceptors vs. Tower middleware

	Tower middleware	Interceptor
Operates on	`http::Request` / `http::Response`	Decoded RPC: `Spec`, headers, deadline, `Payload`
Runs	Before the body is read	`intercept_head`: before the body is read. `intercept_unary` and `intercept_streaming`: after the body is read, before it is decoded (details)
Sees the RPC method	No (must re-parse the URI)	Yes (`head.path()`, `head.spec()`, `ctx.path()`, `ctx.spec()`)
Sees the message body	Compressed/enveloped wire bytes	Lazily decoded, codec-aware `Payload`
Short-circuits	By returning an `http::Response`	By returning `Err` or a `UnaryResponse`
Best for	Authentication, gzip, raw header rewriting, generic HTTP concerns	Authorization on the `Spec` and headers (`intercept_head`), RPC-aware tracing, validation, rate limiting

Both compose: a Tower layer wraps the whole ConnectRpcService (including its interceptor chain). An interceptor that needs an HTTP-level fact (e.g. the remote socket address) reads it from ctx.extensions() after a Tower layer inserts it.

Rejecting before the body is read

Interceptor::intercept_head receives a RequestHead — the path, the resolved Spec, the headers, the protocol, and the request extensions — before the server reads any of the body. It returns Ok(()) to continue or an error to reject. The first rejection ends the request: later interceptors' head checks do not run, no interceptor's intercept_unary or intercept_streaming runs (so count or trace rejections inside the head check or in a Tower layer), the body is not read, and the client gets the error in its protocol's format. As with a Tower layer that returns a response without calling the service, an HTTP/2 stream is reset and an HTTP/1.x connection is usually closed instead of being drained, so a client that is still uploading can see a transport error instead of the error you returned.

A head check can insert values into head.extensions_mut(); later interceptors and the handler read them from ctx.extensions(). That makes it the place to authenticate once and pass the caller on:

use connectrpc::{ConnectError, Interceptor, RequestHead};

struct Caller(String);

struct Authenticate {
    verifier: Verifier,
}

#[connectrpc::async_trait]
impl Interceptor for Authenticate {
    async fn intercept_head(&self, head: &mut RequestHead<'_>) -> Result<(), ConnectError> {
        let token = head.header("authorization").and_then(|v| v.to_str().ok());
        let Some(caller) = token.and_then(|t| self.verifier.verify(t)) else {
            return Err(ConnectError::unauthenticated("missing or invalid token"));
        };
        head.extensions_mut().insert(Caller(caller));
        Ok(())
    }
}

verifier.verify stands for your own credential check; compare secrets with a constant-time comparison, not ==.

Every interceptor's head check runs before any interceptor's intercept_unary or intercept_streaming, so a head check sees the headers as they reached the service, not as an outer interceptor rewrote them with headers_mut. Decide one policy in one hook rather than splitting it across the two. The check also runs for a path that matches no method, with head.spec() None, ahead of the not-found error; gate on head.path() or treat None as a denial. head.peer_addr() (and head.peer_certs() with the server-tls feature) return the connection's address and client certificates when the transport recorded them.

The check has no message to look at, and it runs before the request deadline applies, so wrap a wait on a slow dependency in tokio::time::timeout. A request the service rejects for an unsupported HTTP method or an unrecognized content type never reaches interceptors, and so never reaches this check either; the service drains such a request's body, bounded by max_request_body_size and the request deadline, before it answers.

Authentication and the cost of an unauthenticated request

A credential check should run as early as the server allows, because everything the server does before rejecting a request is work an unauthenticated peer can make it do for free. Tower middleware and intercept_head run before the body is read; intercept_unary and intercept_streaming run after. The difference is what has been spent by the time each one can say no:

Before the hook runs	Tower middleware, `intercept_head`	`intercept_unary`, `intercept_streaming`
Request head parsed	yes	yes
Request body read	no	unary and server-streaming: yes, up to `max_request_body_size` (4 MB by default); client- and bidi-streaming: at most a message or two are read ahead
Body decompressed	no	yes, up to `max_message_size` (4 MB) per message
Message decoded	no	no — `Payload` decodes lazily, only if something reads it

The message decode is the step that multiplies memory: a few bytes of repeated empty messages on the wire become a heap allocation per element, so a 4 MB body can decode into hundreds of MB. In this crate that step happens after the interceptor chain, and is bounded by Limits::element_memory_limit (32 MiB by default) when it does. An intercept_unary or intercept_streaming that returns unauthenticated therefore costs at most the bounded body read and inflate; a Tower layer or intercept_head that does the same costs only the head. Under the default limits, the worst an unauthenticated caller can hold open against a check in intercept_unary or intercept_streaming is max_request_body_size + max_message_size per in-flight request (the compressed body is held while it inflates), multiplied by the HTTP/2 concurrent-stream limit per connection.

Put authentication — "does this caller hold any credential at all" — in Tower middleware or in intercept_head, where it runs before a single body byte is read. The middleware example does this with axum::middleware::from_fn for a bearer token. Put authorization — "may this identity call this method" — in intercept_head too when it needs only the resolved Spec, the headers, and the connection's extensions: a rejection there costs the peer only the head. An authorization check that must read the message runs in intercept_unary or intercept_streaming, where an Err costs the peer the bounded reads above and nothing more; keep max_message_size and max_request_body_size at values the deployment can absorb across its concurrent-stream budget, and reject before touching req.payload.

Hosting

With axum (recommended)

Router::into_axum_service() returns a tower service you mount via axum::Router::fallback_service, and into_axum_router() returns a ready-to-merge axum router. This is the common path because it lets you compose connect RPC routes with regular HTTP routes (health checks, static files, OAuth callbacks):

let app = axum::Router::new()
    .route("/health", axum::routing::get(|| async { "OK" }))
    .fallback_service(connect_router.into_axum_service())
    .layer(/* tower layers */);

let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
axum::serve(listener, app).await?;

Standalone server

Enable the server feature for a built-in hyper-based server. This is the no-frills path when you don't need axum's routing or per-route configuration:

use connectrpc::Server;

let connect_router = Router::new().add_service(service);
Server::new(connect_router)
    .serve("127.0.0.1:8080".parse()?)
    .await?;

The standalone Server handles HTTP/1.1, HTTP/2 with prior knowledge, and graceful shutdown. It's a single dispatcher with no per-route configuration, so add things like health endpoints either as RPC methods or by mounting the Connect service in axum.

Every per-connection setting (with_max_connection_age, the HTTP/2 knobs, ...) lives on ConnectionConfig; Server, BoundServer and connectrpc::axum::serve all accept one via with_connection_config, so a configuration built once applies to any of them.

Custom accept loops

When you need a policy at accept time or per connection that the built-in loop does not have — admit or refuse by client certificate or source address, cap connections per tenant, shed load before HTTP is spoken, serve some clients on a different runtime, listen on a Unix socket — write the loop yourself and hand each accepted stream to Server::serve_connection (or the free server::serve_connection, which takes any tower HTTP service such as an axum::Router). You decide which connections are served and where; the connection driver still gives each one the full lifecycle (settings, timeouts, retirement, GOAWAY on shutdown, panic isolation, PeerAddr / PeerCerts / extensions) and tells you why it ended.

use connectrpc::{CloseReason, ConnectionInfo, Server};

let server = Arc::new(Server::new(router).with_max_connection_age(Duration::from_secs(600)));
let tls = tokio_rustls::TlsAcceptor::from(tls_config);
let listener = tokio::net::TcpListener::bind("0.0.0.0:8443").await?;
let (drain_tx, drain_rx) = tokio::sync::watch::channel(false);
let mut connections = tokio::task::JoinSet::new();
let mut shutdown = std::pin::pin!(tokio::signal::ctrl_c());

loop {
    let (stream, peer) = tokio::select! {
        biased; // a pending shutdown wins over one more accept
        _ = &mut shutdown => break,
        Some(_) = connections.join_next(), if !connections.is_empty() => continue,
        // Production: skip WouldBlock, Interrupted, ConnectionAborted and
        // ConnectionReset, and pause after EMFILE / ENFILE, as Server does.
        accepted = listener.accept() => accepted?,
    };
    // As the built-in loop does: no Nagle delay on small HTTP/2 frames.
    if let Err(err) = stream.set_nodelay(true) {
        tracing::warn!(%peer, %err, "set_nodelay failed");
    }
    // Where this connection runs: say, one `tokio::runtime::Handle` per class
    // of client.
    let runtime = runtime_for(peer);
    // Take the socket off this runtime's I/O driver; the task re-registers it.
    let stream = match stream.into_std() {
        Ok(stream) => stream,
        Err(err) => {
            tracing::warn!(%peer, %err, "detaching the socket failed");
            continue;
        }
    };
    let (server, tls, mut drain) = (Arc::clone(&server), tls.clone(), drain_rx.clone());
    connections.spawn_on(async move {
        let stream = match tokio::net::TcpStream::from_std(stream) {
            Ok(stream) => stream,
            Err(err) => {
                tracing::warn!(%peer, %err, "registering the socket failed");
                return;
            }
        };
        // TLS handshake on the connection's task, not the loop's, with a bound.
        let Ok(Ok(stream)) =
            tokio::time::timeout(Duration::from_secs(10), tls.accept(stream)).await
        else { return };
        let certs: Arc<[_]> = match stream.get_ref().1.peer_certificates() {
            Some(chain) => chain.iter().map(|c| c.clone().into_owned()).collect(),
            None => return, // your policy: no client certificate, no service
        };
        let mut info = ConnectionInfo::new().with_peer_addr(peer).with_peer_certs(certs);
        // Anything handlers should see, computed once per connection.
        let Some(identity) = Identity::from_certs(info.peer_certs()) else { return };
        info.extensions_mut().insert(identity);
        // Everything Server::serve does for a connection, on this task.
        let closed = server
            .serve_connection(stream, info, async move { let _ = drain.wait_for(|d| *d).await; })
            .await;
        // Why it ended, and whether it failed (also while draining). An
        // expired header-read timeout, routine for idle keep-alive
        // connections, carries an error but is not a failure.
        let failed = closed.error().is_some()
            && closed.reason() != CloseReason::HeaderReadTimeout;
        metrics::connection_closed(closed.reason(), failed);
    }, &runtime);
}
drop(listener);                   // refuse new connections
let _ = drain_tx.send(true);      // GOAWAY every live one
while connections.join_next().await.is_some() {}   // wait for them

Serving a connection on another runtime takes the two steps shown. First, spawn its task there with JoinSet::spawn_on: the future binds timers, hyper's per-stream tasks and every handler to the runtime that polls it, and the set can still abort and drain the task. That runtime needs I/O and time enabled (enable_all()); from_std panics without I/O. Second, re-register the socket there: into_std on the accepting side, TcpStream::from_std inside the task. A tokio socket stays on the I/O driver of the runtime that created it, so without this step a busy accepting runtime delays the connection, and shutting that runtime down breaks it.

A TlsStream cannot be rebuilt around a moved socket, so the loop moves the plain TcpStream first and picks the runtime from the peer address, before the TLS handshake. To place clients by class without inspecting certificates, give each class its own listener, with its accept loop on that class's runtime. To place them by client certificate, wrap the TcpStream in your own AsyncRead + AsyncWrite type that can detach the socket (into_std) and re-attach it (from_std), and hand that type to the TLS acceptor. After the handshake, reach the socket through TlsStream::get_mut: detach it while the accepting runtime is still running, re-attach it inside the task on the target runtime, and don't poll the stream in between.

The loop must also outlive what it spawned: track the tasks and drain them before returning, as the code above shows.

Raw hyper

For hyper connection-builder settings ConnectionConfig does not expose (HPACK table size, max frame size, ...), ConnectRpcService is an ordinary tower service: wrap it in hyper_util::service::TowerToHyperService and hand it to hyper_util::server::conn::auto::Builder yourself.

use connectrpc::{ConnectRpcService, Router};
use hyper_util::{
    rt::{TokioExecutor, TokioIo},
    server::conn::auto::Builder as AutoBuilder,
    service::TowerToHyperService,
};

let connect_service = ConnectRpcService::new(Router::new().add_service(greeter_service));
let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
let mut builder = AutoBuilder::new(TokioExecutor::new());
builder.http2().max_frame_size(1 << 20);

loop {
    let (stream, _peer_addr) = listener.accept().await?;
    let conn = builder
        .serve_connection(TokioIo::new(stream), TowerToHyperService::new(connect_service.clone()))
        .into_owned();
    tokio::spawn(async move {
        if let Err(err) = conn.await {
            eprintln!("connection ended with error: {err}");
        }
    });
}

You then own the whole connection lifecycle — none of the timeouts, retirement, graceful shutdown, panic isolation or PeerAddr / PeerCerts stamping described above happens unless you re-implement it — so prefer a custom accept loop built on serve_connection over raw hyper unless it is specifically a hyper knob you are missing.

TLS

Enable the server-tls feature (or the tls umbrella feature for both server and client TLS).

For the standalone Server:

use std::sync::Arc;

let server_config: Arc<rustls::ServerConfig> = /* load PEMs, build config */;

Server::new(connect_router)
    .with_tls(server_config)
    .serve("0.0.0.0:8443".parse()?)
    .await?;

For the axum path, connectrpc::axum::serve_tls (requires both the axum and server-tls features; plaintext connectrpc::axum::serve needs axum and server) is a drop-in replacement for axum::serve that runs the app on the same accept loop and connection driver as the standalone Server — TLS termination, PeerAddr / PeerCerts in request extensions, every ConnectionConfig setting, graceful GOAWAY, panic isolation — so handler code that reads ctx.peer_certs() is portable across both hosting paths:

let app = axum::Router::new()
    .route("/health", axum::routing::get(|| async { "OK" }))
    .fallback_service(connect_router.into_axum_service());

let listener = tokio::net::TcpListener::bind("0.0.0.0:8443").await?;
connectrpc::axum::serve_tls(listener, app, server_config)
    .with_graceful_shutdown(shutdown_signal)
    .await?;

Work that depends only on the connection — parsing a workload identity out of the client certificate, say — need not repeat on every request. with_connection_extensions (on Server, BoundServer and connectrpc::axum::Serve) registers a function that runs once per accepted connection, after the TLS handshake and before the first request. It reads the connection's ConnectionInfo (peer_addr(), peer_certs(), extensions()) and inserts into the http::Extensions it is handed. That map starts empty; what the function inserts joins the connection's extensions, replacing entries of the same type, is cloned into every request on that connection, and is read with ctx.extensions().get::<T>(). The function cannot remove an entry or change the peer: whatever it inserts under PeerAddr / PeerCerts, requests get those types from ConnectionInfo's peer fields, which the built-in loops set from the transport. Inserted values must be Clone + Send + Sync + 'static; wrap large ones in an Arc.

Server::serve_connection runs the function too, after whatever a custom accept loop put in info.extensions_mut(); a loop around the free server::serve_connection writes info.extensions_mut() itself.

#[derive(Clone)]
struct PeerIdentity(Arc<str>);

Server::new(connect_router)
    .with_tls(server_config)
    .with_connection_extensions(|conn, ext| {
        if let Some(id) = parse_identity(conn.peer_certs()) {
            ext.insert(PeerIdentity(id));
        }
    })
    .serve("0.0.0.0:8443".parse()?)
    .await?;

// In a handler:
let who = ctx.extensions().get::<PeerIdentity>();

The eliza example (examples/eliza/README.md) walks through generating self-signed certificates with openssl, configuring mTLS via --client-ca, and the rustls strict-PKI requirement that your CA cert must be distinct from the server leaf cert. The mtls-identity example (examples/mtls-identity/README.md) demonstrates serve_tls end-to-end with cert-SAN identity parsed once per connection and an ACL keyed on it.

Health checking

The connectrpc-health crate implements the standard grpc.health.v1.Health service. Mount it on your Connect router and clients like grpc_health_probe, kubelet's grpc: probe, and gRPC-aware service meshes (Linkerd, Istio) just work.

This is the gRPC protocol — different from the plain HTTP GET /health route shown earlier in the Hosting section. Keep the HTTP route for httpGet: probes; add the gRPC service for grpc: probes.

[dependencies]
connectrpc = { version = "0.9", features = ["server"] }
connectrpc-health = "0.9"

use connectrpc::Router;
use connectrpc_health::{install_static, Status};

// `install_static` registers every name with `Status::Serving`; use the
// generated `*_SERVICE_NAME` constants from your service stubs so the
// registered name matches exactly what clients ask for. The
// whole-process `""` entry is seeded for you, so probes that don't
// pass a service name also work.
let (router, health) = install_static(Router::new(), [
    proto::greet::v1::GREET_SERVICE_SERVICE_NAME,
]);

// Flip status when something goes wrong. `set_status` errors on an
// unknown name, so typos surface immediately instead of silently
// shadowing the real entry.
health
    .set_status(proto::greet::v1::GREET_SERVICE_SERVICE_NAME, Status::NotServing)
    .expect("registered above");

// At shutdown, drain. `shutdown()` flips every registered service,
// including the empty whole-process entry:
health.shutdown();

For custom logic (e.g. report NotServing while a database connection is down), implement the Checker trait directly and wrap it in HealthService::new(...) or HealthService::from_arc(...). The default Checker::watch body returns Unimplemented, which is fine for Check-only probes; override it if your probes call Watch.

The health routes accept at most 16 KiB per request. A HealthCheckRequest is one service name, so install_static sizes Check and Watch to connectrpc_health::request_limits() — a per-route Limits profile (see Request limits) that replaces the service-wide limits on those two routes, whether those are looser or tighter — and a larger request is refused with resource_exhausted before it reaches the checker. The profile also holds the decode budget to four times the 16 KiB message bound (64 KiB), and StaticChecker's not_found error echoes at most 128 bytes of an unregistered name, so the error message is bounded by a constant rather than by the size of the request (a custom Checker bounds its own error text). Registering a HealthService any other way (HealthExt::register, Router::add_service) does not apply it, so follow that with apply_request_limits(router, request_limits()). To tune the health routes specifically, call apply_request_limits with your own Limits after either path; the later call wins:

use connectrpc_health::{apply_request_limits, install_static, request_limits};

let (router, health) = install_static(Router::new(), [/* ... */]);
// Start from `request_limits()` so the rest of the profile carries over.
let router = apply_request_limits(
    router,
    request_limits()
        .with_max_request_body_size(1024)
        .with_max_message_size(1024),
);

The HealthClient (for in-process probes, integration tests, sidecar tooling) is gated on a client Cargo feature that is on by default. Server-only deployments turn it off:

[dependencies]
connectrpc = { version = "0.9", features = ["server"] }
connectrpc-health = { version = "0.9", default-features = false }

That drops connectrpc/client (the HTTP/2 transport stack) from the dependency graph entirely. use connectrpc_health::HealthClient then becomes an unresolved import, but the binary stays lean.

Unknown services on Watch. Non-empty unregistered services return Err(ConnectError::not_found(_)) from both Check and Watch; the empty service auto-subscribes on Watch and returns Serving on Check by default. The gRPC Health spec additionally describes a SERVICE_UNKNOWN keep-stream-open flow for Watch that this crate does not implement, matching the Go connectrpc.com/grpchealth reference. Every probe that treats any error as a failure — kubelet's grpc: probe, grpc_health_probe, Linkerd, Istio — works unchanged. See HealthService's # Unknown services section in the crate docs for the full context.

Server reflection

The connectrpc-reflection crate implements the standard gRPC server reflection service (grpc.reflection.v1 and its v1alpha predecessor), so schema-aware clients — grpcurl, buf curl, Postman, grpcui — can discover and call your services without local proto files, over gRPC, gRPC-Web, and the Connect protocol alike.

[dependencies]
connectrpc = { version = "0.9", features = ["server"] }
connectrpc-reflection = "0.9"

Emit a descriptor set from your build script (see Code generation), embed it, and mount the service:

// build.rs: .emit_descriptor_set("app.fds.bin") before .compile()
use connectrpc::Router;
use connectrpc_reflection::{Reflector, install};

let bytes: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/app.fds.bin"));
let reflector = Reflector::from_descriptor_set_bytes(bytes)?;
// `router` is your service router from `register()`.
let router = install(router, reflector); // mounts v1 + v1alpha

The reflection routes accept at most 16 KiB per request message. A ServerReflectionRequest is a host plus one symbol or file name, so install sizes both versions' routes to connectrpc_reflection::request_limits() — a per-route Limits profile (see Request limits) that replaces the service-wide limits on those routes, whether those are looser or tighter; the RPC is a bidirectional stream, so the bound is per message rather than per call. The profile also holds the per-message decode budget to four times the 16 KiB bound (64 KiB), and a lookup miss echoes at most 128 bytes of the queried name in its ErrorResponse, so the error text is bounded by a constant rather than by the size of the request; the protocol's own original_request and valid_host echoes remain, so a response is at most about twice the request that produced it. Mounting a single version through the generated ServerReflectionExt::register (or using Router::add_service) does not apply it, so follow that with apply_request_limits(router, request_limits()). To tune the reflection routes specifically, call apply_request_limits with your own Limits after either path; the later call wins:

use connectrpc_reflection::{apply_request_limits, install, request_limits};

let router = install(router, reflector);
// Start from `request_limits()` so the rest of the profile carries over.
let router = apply_request_limits(router, request_limits().with_max_message_size(1024));

Alternatively, when your buffa codegen has reflection enabled, serve straight from the generated package's descriptor pool with Reflector::from_descriptor_pool(proto::descriptor_pool().clone()) — no build-script step. The bytes path answers with the compiler's original per-file descriptor bytes; the pool path re-encodes (semantically faithful, unknown fields preserved). See the Reflector crate docs for the trade-off.

Reflection intentionally publishes your schema. Everything in the descriptor set is exposed — all files, their transitive imports, and every compiled service, whether or not its handlers are mounted. Gate or omit the service on deployments where that is not wanted.

Two more behaviors worth knowing before deploying:

Reflector::with_services curates the advertised list (the override is verbatim, like Go grpcreflect's Namer) — use it when the descriptor set compiles in more services than you want to advertise; Reflector::service_names inspects the current list.
The service is self-describing: queries about grpc.reflection.* fall back to the crate's own descriptors and ListServices includes the reflection services, matching grpc-go. Schema-free callers like buf curl need this to invoke ServerReflectionInfo at all.

The multiservice example mounts reflection with both descriptor sources (selectable via REFLECTION_SOURCE=fds|pool), and reflection-demo.sh walks through discovery and schema-free calls with buf curl.

Production hardening

Deadline policy

Connect and gRPC clients send a per-request timeout header (Connect-Timeout-Ms or grpc-timeout). With no policy, the server trusts that value verbatim: a Connect-Timeout-Ms: 1 request cancels the handler mid-write, while a Connect-Timeout-Ms: 86400000 request holds a worker for 24 hours, and a request with no timeout header runs unbounded. DeadlinePolicy gives the server the say.

use connectrpc::{ConnectRpcService, DeadlinePolicy};
use std::time::Duration;

let policy = DeadlinePolicy::new()
    .with_min(Duration::from_millis(5))           // floor: reject "cancel me instantly"
    .with_max(Duration::from_secs(30))            // cap: bound worker lifetime
    .with_default_timeout(Duration::from_secs(10)) // applied when client asserts nothing
    .with_enforce_on_streams(true);               // also cut off streaming bodies

let service = ConnectRpcService::new(router)
    .with_deadline_policy(policy);
// or: Server::new(router).with_deadline_policy(policy)

Why each knob:

with_max is the most important one for any service that accepts untrusted callers — without it a client controls how long a worker stays busy. Set it to your longest acceptable request — for unary and server-streaming RPCs the capped budget covers receiving the request body as well as handler execution, so size it for uploads, not just handler runtime.
with_default_timeout matters because the timeout header is optional. A request that omits it has no bound at all unless you set one. Set it to your SLA. For unary and server-streaming RPCs, the budget includes receiving the request body as well as handler execution.
with_min protects against a misbehaving or adversarial client cancelling the handler before it can do anything (e.g. mid-write on a streaming response). A few milliseconds is usually enough.
with_enforce_on_streams(true) closes the streaming-body gap. By default the deadline only bounds the time-to-first-response — once a server- or bidi-streaming handler returns its stream, the items flow unbounded. Enabling this wraps the response body so the next item after the deadline is a deadline_exceeded error and the stream ends. Cancellation drops the inner stream at the next yield point with no grace period; spawn commit-critical work off the request future if it must outlive the caller.
with_inter_message_timeout(d) detects stalled streams (a handler waiting on a slow upstream). Independent of with_enforce_on_streams — takes effect whenever set, with or without the absolute deadline. Arms when the response stream is first polled (stream-setup latency before that is not counted) and resets on each yielded item.

DeadlinePolicy::new() with no with_* calls is a no-op that preserves the prior default behavior. Existing services see no change without opting in.

When a client value is clamped, a tracing::debug! event fires on target connectrpc::deadline with the path and before/after durations. Enable RUST_LOG=connectrpc::deadline=debug to spot misbehaving clients.

Inside a handler, ctx.deadline() reflects the moderated value (after clamping), so the handler can budget downstream calls — propagate the remaining time minus a margin as the timeout for outbound RPCs. ctx.time_remaining() does the subtraction for you (None when the request has no deadline):

if let Some(remaining) = ctx.time_remaining() {
    options = options.with_timeout(remaining.saturating_sub(margin));
}

Request limits

Limits bounds what a request may cost before a handler sees it: the body as read from the socket (with_max_request_body_size, 4 MB), each message after decompression (with_max_message_size, 4 MB), and the memory a decode may commit to repeated and string elements (with_element_memory_limit, 32 MiB). The body and message limits are enforced while reading and inflating, so an oversized or highly compressed request is refused with resource_exhausted without ever being fully buffered. Set them service-wide on ConnectRpcService:

let service = ConnectRpcService::new(router)
    .with_limits(Limits::default().with_max_message_size(1024 * 1024));

A single route can carry its own Limits with Router::with_route_limits, and those replace the service-wide ones for that method — every field, not a field-by-field minimum. Use it to size a method to its actual request profile: one whose legitimate requests are a few hundred bytes need not accept the service default, and one upload-shaped method can exceed the default without raising it for everything else.

let router = service
    .register(Router::new())
    .with_route_limits(
        PING_SERVICE_PING_SPEC.procedure,
        Limits::default()
            .with_max_request_body_size(16 * 1024)
            .with_max_message_size(16 * 1024),
    );

Per-route limits are a Router feature; the generated monomorphic FooServiceServer<T> dispatchers use the service-wide limits.

The bundled connectrpc-health and connectrpc-reflection services use this mechanism on their own routes: install_static and install apply a 16 KiB-per-message profile (request_limits()), and each crate's apply_request_limits(router, limits) re-applies or tunes it — see Health checking and Server reflection.

Clients

Enable the client feature for HTTP client support with connection pooling.

HttpClient

HttpClient is the standard transport built on hyper. Construct one of two variants: cleartext (http:// only) or TLS-enabled (https:// only):

use connectrpc::client::HttpClient;

// Cleartext
let http = HttpClient::plaintext();

// TLS - requires client-tls or tls feature
let tls_config: Arc<rustls::ClientConfig> = /* trust store + ALPN */;
let http = HttpClient::with_tls(tls_config);

A plaintext() client refuses https:// URIs and a with_tls() client refuses http:// URIs - this catches misconfiguration loudly rather than silently downgrading.

Connection-establishment bounds and keep-alive

Both HttpClient and Http2Connection bound connection establishment by default: a 20-second wall-clock budget on the whole DNS + TCP + TLS chain (DEFAULT_ESTABLISHMENT_TIMEOUT) with an additional 5-second per-address TCP bound (DEFAULT_TCP_CONNECT_TIMEOUT). Exceeding either surfaces as ErrorCode::Unavailable, so a server that accepts the TCP connection but stalls the TLS handshake cannot park poll_ready indefinitely. To adjust or opt out, use the builder() entry point on either transport:

use std::time::Duration;
use connectrpc::client::Http2Connection;

let conn = Http2Connection::builder()
    .establishment_timeout(Duration::from_secs(10))
    .keep_alive_interval(Duration::from_secs(30))
    .keep_alive_while_idle(true)
    .connect_tls(uri, tls_config)
    .await?;

Http2ConnectionBuilder also proxies hyper's HTTP/2 keep-alive and flow-control knobs (keep_alive_interval, keep_alive_timeout, keep_alive_while_idle, initial_stream_window_size, initial_connection_window_size, adaptive_window), with a TokioTimer pre-wired so the keep-alive setters work without further plumbing. The h2_settings(|b| ...) escape hatch exposes the underlying hyper builder for knobs not surfaced directly. To restore the unbounded pre-0.8.0 behaviour, chain .no_establishment_timeout().no_tcp_connect_timeout().

On a multi-homed host, Http2ConnectionBuilder::local_address(IpAddr) binds the built-in connector's socket to one of the host's addresses before connecting, so the connection (and every reconnect) originates from that address — useful when the peer keys on the source address it observes, or when egress must leave a particular interface. The resolved peer addresses are filtered to that address's family; a peer with no address of that family fails to connect rather than connecting from a kernel-chosen source. Like tcp_connect_timeout, it applies only to the built-in TCP connector, not to the custom-connector / Unix-socket terminals. (HttpClientBuilder does not expose this; use Http2Connection when you need a pinned source address.)

ClientConfig

ClientConfig carries the base URI and per-call defaults that apply to every RPC made with the client:

use std::time::Duration;
use connectrpc::client::ClientConfig;

let config = ClientConfig::new("http://localhost:8080".parse()?)
    .with_default_timeout(Duration::from_secs(30))
    .with_default_header("authorization", "Bearer demo-token")
    .with_default_header("x-trace-id", "trace-12345");

These defaults automatically apply to every call from that client. Use them for cross-cutting concerns like auth or tracing IDs.

CallOptions

For per-call overrides, use the _with_options method variants and pass CallOptions:

use connectrpc::client::CallOptions;

let resp = client.greet_with_options(
    GreetRequest { name: "World".into(), ..Default::default() },
    CallOptions::default()
        .with_timeout(Duration::from_secs(5))
        .with_max_message_size(1024 * 1024),
).await?;

Per-call options replace config defaults for the fields they set (timeout here); other defaults (the auth header) still apply.

Reading the response

Unary responses give you several access patterns:

let resp = client.greet(req).await?;

// Pattern 1: borrow the view via .view(). Zero-copy. Field access
// (.greeting -> &str) works directly on the returned view, and the
// response handle keeps headers/trailers available alongside it.
println!("{}", resp.view().greeting);
let _ = resp.headers();
let _ = resp.trailers();

// Pattern 2: consume via .into_view() to get the OwnedView. Still
// zero-copy - read fields through .reborrow() - but discards
// headers/trailers.
let msg = client.greet(req).await?.into_view();
let greeting: &str = msg.reborrow().greeting;

// Pattern 3: .into_owned() for the prost-style owned struct.
// Allocates and copies all string/bytes fields.
let owned: GreetResponse = client.greet(req).await?.into_owned();

// Pattern 4: .into_owned_parts() when you need the owned struct AND
// the response metadata - the metadata-preserving form of Pattern 3.
let (headers, owned, trailers) = client.greet(req).await?.into_owned_parts();

Custom transports

Generated clients are generic over ClientTransport, which is auto- implemented for any tower::Service that handles http::Request<ClientBody> and returns http::Response<B>. So you can plug in any tower stack as the transport:

use tower::ServiceBuilder;
use tower_http::timeout::TimeoutLayer;
use connectrpc::client::{Http2Connection, ServiceTransport};

let conn = Http2Connection::connect_plaintext(uri).await?.shared(1024);
let stacked = ServiceBuilder::new()
    .layer(TimeoutLayer::new(Duration::from_secs(30)))
    .service(conn);

let client = GreetServiceClient::new(
    ServiceTransport::new(stacked),
    config,
);

This is also how the wasm example (examples/wasm-client/) plugs in a browser fetch-based transport.

A hand-written ClientTransport must give its ResponseBody an error type that converts into Box<dyn std::error::Error + Send + Sync> — any std::error::Error + Send + Sync + 'static type, or that boxed type itself; the call functions and the generated clients require it so that a failure while reading the body can be kept as the surfaced error's source() (see Errors and status codes). If the body you wrap reports a Display-only error, implement Error for it or adapt the body with http_body_util::BodyExt::map_err.

Errors and status codes

ConnectError is the error type for both server-returned and client-observed errors:

pub struct ConnectError {
    pub code: ErrorCode,
    pub message: Option<String>,
    pub details: Vec<ErrorDetail>,
    // response headers and trailers: private, exposed via the
    // response_headers()/trailers() accessors and their _mut variants
    // source: private, exposed via std::error::Error::source() and
    // source_arc(); never serialized
}

ErrorCode is the canonical Connect/gRPC status set: Canceled, Unknown, InvalidArgument, DeadlineExceeded, NotFound, AlreadyExists, PermissionDenied, ResourceExhausted, FailedPrecondition, Aborted, OutOfRange, Unimplemented, Internal, Unavailable, DataLoss, Unauthenticated.

Construct one with the message:

return Err(ConnectError::new(
    ErrorCode::PermissionDenied,
    format!("user {user} cannot read {name}"),
));

The dispatcher maps each code to the appropriate HTTP status (e.g. NotFound -> 404, Unauthenticated -> 401, PermissionDenied -> 403) and the appropriate protocol-specific representation. Clients parse it back into the same ConnectError shape regardless of which protocol they're speaking.

For more structured errors, attach ErrorDetail entries (which carry typed protobuf messages) before returning. These flow through to clients in the standard Connect error-detail wire format.

To keep an underlying error's cause available for logging without sending it to the client, attach it with .with_source(err). It is surfaced through std::error::Error::source() (bring the trait into scope, or call std::error::Error::source(&err)) and, as a SharedSource handle that can be moved into another error type, through err.source_arc(); it is never serialized. A ConnectError decoded from a server's response therefore always has .source().is_none(), even when its message is populated; only an error that code in this process attached a cause to carries one.

On the client, a source is present whenever the failure came from the transport, whether it happened before or after the response headers arrived (provided the transport followed the convention at the end of this section), and absent when the call path synthesised the error itself. The built-in transports attach the underlying hyper / rustls / std::io error for DNS resolution, connection refused, TLS handshake, HTTP/2 connection establishment (including its timeout) and request send; a failure while reading the response body — a stream reset mid-body, say — is attached by the call path on every RPC shape, which is why the public call functions and the generated clients require the transport's body error type to convert into Box<dyn std::error::Error + Send + Sync> (any std::error::Error + Send + Sync + 'static type does, and so does that boxed type itself). The one exception is a non-2xx gRPC response whose body dies while being read for a grpc-status: the HTTP status is the error reported, and the read failure is dropped rather than attached. Either way err.source() yields the error the transport reported, as its own type: for the built-in transports that is a hyper / hyper-util error whose own source() chain leads on to the underlying io::Error or h2 reset, so walking the chain (or the transport's error directly, for a custom transport that reports io::Error) reaches an io::ErrorKind to inspect. Only that error's Display text reaches message, and it is repeated there deliberately so that plain {} formatting stays informative; a renderer that also walks the source chain, such as anyhow's {:#}, will print it twice. Errors with no transport cause behind them do not carry a source: the call deadline (with_timeout / with_default_timeout) when the local timer fires, request construction and encoding failures, and response decoding. A body read that fails after the deadline has passed is reported as deadline_exceeded but keeps the read error as its source, since the reset that ended the stream is still the most specific fact about the failure. A custom ClientTransport that returns its own ConnectError from send should call .with_source(..) itself, or build the error with ConnectError::unavailable_from_transport, to follow the same convention; a ConnectError its body reports is not surfaced verbatim the way one from send is, but is attached as the source of an internal (or deadline_exceeded) error.

Compression

The runtime ships with gzip, zstd, and identity by default. Servers advertise supported algorithms in the accept-encoding response and honor the client's connect-content-encoding request header (or grpc-encoding for the gRPC protocols).

Per-RPC compression control

A handler can override the server's compression policy for a single response via Response::compress:

async fn greet(
    &self,
    _ctx: RequestContext,
    req: ServiceRequest<'_, GreetRequest>,
) -> ServiceResult<GreetResponse> {
    let mut resp = Response::new(/* ... */);
    if response_is_huge() {
        resp = resp.compress(true);  // force compress this response
    }
    Ok(resp)
}

Custom compression algorithms

CompressionRegistry is pluggable. Implement CompressionProvider for your algorithm and register it on the dispatcher:

use connectrpc::{CompressionProvider, CompressionRegistry, ConnectError};
use bytes::Bytes;

struct MyCompression;

impl CompressionProvider for MyCompression {
    // Sent in content-encoding / accept-encoding, so it must be an HTTP
    // token; `register` panics on anything else.
    fn name(&self) -> &'static str { "my-algo" }

    fn compress(&self, data: &[u8]) -> Result<Bytes, ConnectError> {
        // ...
    }

    fn decompressor<'a>(
        &self,
        data: &'a [u8],
    ) -> Result<Box<dyn std::io::Read + 'a>, ConnectError> {
        // Return a reader that yields decompressed bytes. The framework
        // controls how much is read, so decompression is bounded by
        // Limits::max_message_size.
        // ...
    }
}

let registry = CompressionRegistry::default().register(MyCompression);
let service = ConnectRpcService::new(router).with_compression(registry);

Examples directory tour

Example	What it covers
`streaming-tour/`	All four RPC types (unary, server stream, client stream, bidi) on a trivial NumberService. Smallest demo of handler signatures and client invocation patterns.
`middleware/`	Server-side tower middleware composition: an `axum::middleware::from_fn` bearer-token auth, identity passthrough via `RequestContext::extensions()`, response trailers via `Response::with_trailer`. Client demos `ClientConfig::with_default_header` and `CallOptions::with_timeout`.
`mtls-identity/`	mTLS twin of `middleware/`: axum hosted behind `connectrpc::axum::serve_tls`, identity parsed from the client cert's DNS SAN once per connection with `with_connection_extensions` instead of a bearer token, ACL keyed on the cert-derived identity. In-memory `rcgen` PKI; no PEM files.
`eliza/`	Production-shaped streaming app: a port of the `connectrpc/examples-go` ELIZA demo. Server-streaming Introduce + bidi-streaming Converse, TLS, mTLS, CORS, IPv6, both server and client binaries, interoperates with the hosted Go reference at `demo.connectrpc.com`.
`multiservice/`	Multiple proto packages compiled together with `buf generate`, multiple services on one server, well-known type usage, and server reflection mounted from both descriptor sources (`REFLECTION_SOURCE=fds\|pool`; see `reflection-demo.sh`).
`wasm-client/`	Browser fetch transport: same generated client used from `wasm32-unknown-unknown` with a custom `ClientTransport` backed by `web-sys::fetch`.
`bazel/`	Bazel build integration via custom rules.

Most examples have their own README with run instructions; the rest document themselves through their test.sh / demo scripts.

FilesExpand file tree

guide.md

Latest commit

History

guide.md

File metadata and controls

connectrpc User Guide

Contents

Installation

MSRV

Feature flags

Proto-only (no-JSON) builds

Quick start

Code generation

connectrpc-build (build-time, simplest)

buf generate (checked-in code, production-grade)

Inclusion patterns side-by-side

Very large schemas

Implementing servers

Handler signatures

RequestContext and Response

What you see vs. what you write

The refining_impl_trait lint

Returning a view body

Returning errors

Registering services on a Router

Testing handlers

Streaming RPCs

Server streaming

Client streaming

Bidirectional streaming

Calling streaming RPCs from a client

Tower middleware

Composing layers

Passing data from a layer to a handler

Static method metadata (Spec)

Short-circuit responses

Interceptors

Ordering and registration

Reading and rewriting the request

Short-circuiting and re-running

Streaming RPCs

Interceptors vs. Tower middleware

Rejecting before the body is read

Authentication and the cost of an unauthenticated request

Hosting

With axum (recommended)

Standalone server

Custom accept loops

Raw hyper

TLS

Health checking

Server reflection

Production hardening

Deadline policy

Request limits

Clients

HttpClient

Connection-establishment bounds and keep-alive

ClientConfig

CallOptions

Reading the response

Custom transports

Errors and status codes

Compression

Per-RPC compression control

Custom compression algorithms

Examples directory tour

`connectrpc-build` (build-time, simplest)

`buf generate` (checked-in code, production-grade)

`RequestContext` and `Response`

The `refining_impl_trait` lint

Static method metadata (`Spec`)