connectrpc 0.11.0

Server and client runtime library for Connect RPC

Connect for Python

A Python implementation of Connect: Protobuf RPC that works.

This repo provides a Python implementation of Connect, including both client and server support. It includes a protoc plugin that generates typed client stubs and server interfaces from your .proto files, along with runtime libraries for both synchronous and asynchronous code.

Features

• Clients: Both synchronous and asynchronous clients backed by pyqwest
• Servers: WSGI and ASGI server implementations for use with any Python app server
• Type Safety: Fully type-annotated, including the generated code
• Compression: Built-in support for gzip, brotli, and zstd compression
• Interceptors: Server-side and client-side interceptors for cross-cutting concerns
• Streaming: Full support for server, client, and bidirectional streaming
• Standards Compliant: Verified implementation using the official Connect conformance test suite

Installation

Install the runtime library

pip install connectrpc

Or with your preferred package manager:

# Using uv
uv add connectrpc

# Using poetry
poetry add connectrpc

Install the code generator

With a protobuf definition in hand, you can generate stub code. This is easiest using buf, but you can also use protoc if you're feeling masochistic.

A reasonable buf.gen.yaml:

version: v2
plugins:
  - remote: buf.build/bufbuild/py
    out: gen
  - remote: buf.build/connectrpc/python
    out: gen

Or, you can install the compiler (e.g. pip install protoc-gen-connectrpc), and it can be referenced as protoc-gen-connectrpc. Then, you can use protoc-gen-connectrpc as a local plugin:

- local: .venv/bin/protoc-gen-connectrpc
  out: .

protoc-gen-connectrpc is only needed for code generation. Your actual application should include connectrpc as a dependency for the runtime component.

google.protobuf compatibility

Connect defaults to targeting protobuf-py as the Protocol Buffers implementation but Google's Protocol Buffers for Python are also fully supported. Pass protobuf=google to the codegen plugin to use it.

version: v2
plugins:
  - remote: buf.build/protocolbuffers/python
    out: .
  - remote: buf.build/protocolbuffers/pyi
    out: .
  - remote: buf.build/connectrpc/python
    out: .
    opt: protobuf=google

If configuring a client for JSON codec, make sure to pass connectrpc.compat.google_protobuf_json_codec instead of connectrpc.codec.proto_json_codec.

Basic Client Usage

from your_service_pb import HelloRequest, HelloResponse
from your_service_connect import HelloServiceClient

# Create async client
async def main():
    async with HelloServiceClient("https://api.example.com") as client:
        # Make a unary RPC call
        response = await client.say_hello(HelloRequest(name="World"))
        print(response.message)  # "Hello, World!"

Basic Server Usage

from connectrpc.request import RequestContext
from your_service_pb import HelloRequest, HelloResponse
from your_service_connect import HelloService, HelloServiceASGIApplication

class MyHelloService(HelloService):
    async def say_hello(self, request: HelloRequest, ctx: RequestContext) -> HelloResponse:
        return HelloResponse(message=f"Hello, {request.name}!")

# Create ASGI app
app = HelloServiceASGIApplication(MyHelloService())

# Run with any ASGI server like uvicorn:
# uvicorn server:app --port 8080

Basic Client Usage (Synchronous)

from your_service_pb import HelloRequest
from your_service_connect import HelloServiceClientSync

# Create sync client
def main():
    with HelloServiceClientSync("https://api.example.com") as client:
        # Make a unary RPC call
        response = client.say_hello(HelloRequest(name="World"))
        print(response.message)  # "Hello, World!"

if __name__ == "__main__":
    main()

Check out the docs for more detailed usage including streaming, interceptors, and other advanced features.

Streaming Support

Connect supports all RPC streaming types:

• Unary: Single request, single response
• Server Streaming: Single request, multiple responses
• Client Streaming: Multiple requests, single response
• Bidirectional Streaming: Multiple requests, multiple responses

Server Streaming

Single request, multiple responses:

# Server implementation
async def make_hats(self, req: Size, ctx: RequestContext) -> AsyncIterator[Hat]:
    for i in range(3):
        yield Hat(size=req.inches + i, color=["red", "green", "blue"][i])

# Client usage
async for hat in client.make_hats(Size(inches=12)):
    print(f"Received: {hat}")

Client Streaming

Multiple requests, single response:

# Server implementation
async def collect_sizes(self, reqs: AsyncIterator[Size], ctx: RequestContext) -> Summary:
    total = 0
    count = 0
    async for size in reqs:
        total += size.inches
        count += 1
    return Summary(total=total, average=total/count if count else 0)

# Client usage
async def send_sizes():
    for i in range(5):
        yield Size(inches=i * 2)

summary = await client.collect_sizes(send_sizes())

Bidirectional Streaming

Multiple requests and responses:

# Server implementation (like the Eliza chatbot)
async def converse(self, reqs: AsyncIterator[ConverseRequest], ctx: RequestContext) -> AsyncIterator[ConverseResponse]:
    async for req in reqs:
        # Process and respond to each message
        reply = process_message(req.sentence)
        yield ConverseResponse(sentence=reply)

# Client usage
async def chat():
    yield ConverseRequest(sentence="Hello")
    yield ConverseRequest(sentence="How are you?")

async for response in client.converse(chat()):
    print(f"Response: {response.sentence}")

Streaming Notes

• HTTP/2 ASGI servers: Support all streaming types including full-duplex bidirectional

• HTTP/1.1 servers: Support half-duplex bidirectional streaming only

• WSGI servers: Support streaming but not full-duplex bidirectional due to protocol limitations

• Clients: Support half-duplex bidirectional streaming only

Examples

The example/ directory contains complete working examples demonstrating all features:

• Eliza Chatbot: A Connect implementation of the classic ELIZA psychotherapist chatbot
  • eliza_service.py - Async ASGI server implementation
  • eliza_service_sync.py - Synchronous WSGI server implementation
  • eliza_client.py - Async client example
  • eliza_client_sync.py - Synchronous client example
• All streaming patterns: Unary, server streaming, client streaming, and bidirectional
• Integration examples: Starlette, Flask, and other frameworks

Run the Eliza example:

# Start the server
cd example
uvicorn example.eliza_service:app --port 8080

# In another terminal, run the client
python -m example.eliza_client

Supported Protocols

• ✅ Connect Protocol over HTTP/1.1 and HTTP/2
• ✅ gRPC Protocol support
• ✅ gRPC-Web Protocol support

Server Runtime Options

We verify the following servers with ConnectRPC's conformance suite.

For ASGI servers:

• pyvoy - Fully-featured ASGI server, enables all of Connect's features
• Uvicorn - Lightning-fast ASGI server for HTTP/1

For WSGI servers:

• pyvoy - Fully-featured WSGI server, enables all of Connect's features
• Gunicorn - Python WSGI HTTP Server for HTTP/1

Other ASGI and WSGI servers should also generally work though we have found some issues with flakiness with our conformance tests. If you don't have any preference, we recommend one of the above servers.

WSGI Support

Connect provides full WSGI support via ConnectWSGIApplication for synchronous Python applications. This enables integration with traditional WSGI servers like Gunicorn and uWSGI.

from connectrpc.request import RequestContext
from connectrpc.server import ConnectWSGIApplication
from your_service_pb import Request, Response
from your_service_connect import YourService, YourServiceWSGIApplication

class YourServiceImpl(YourService):
    def your_method(self, request: Request, ctx: RequestContext) -> Response:
        # Synchronous implementation
        return Response(message="Hello from WSGI")

    # WSGI also supports streaming (except full-duplex bidirectional)
    def stream_data(self, request: Request, ctx: RequestContext) -> Iterator[Response]:
        for i in range(3):
            yield Response(message=f"Message {i}")

# Create WSGI application
app = YourServiceWSGIApplication(YourServiceImpl())

# Run with gunicorn: gunicorn server:app

Compression Support

Connect supports multiple compression algorithms:

• gzip: Built-in support, always available
• brotli: Available when brotli package is installed
• zstd: Available when zstandard package is installed

Compression is automatically negotiated between client and server based on the Accept-Encoding and Content-Encoding headers.

Interceptors

Server-Side Interceptors

Interceptors allow you to add cross-cutting concerns like authentication, logging, and metrics:

from connectrpc.interceptor import MetadataInterceptor
from connectrpc.request import RequestContext

class LoggingInterceptor:
    """Implements the MetadataInterceptor protocol."""

    async def on_start(self, ctx: RequestContext) -> None:
        print(f"Handling {ctx.method().name} request")

    async def on_end(self, token: None, ctx: RequestContext, error: Exception | None) -> None:
        if error:
            print(f"Failed {ctx.method().name}: {error}")
        else:
            print(f"Completed {ctx.method().name} request")

# Add to your application
app = HelloServiceASGIApplication(
    MyHelloService(),
    interceptors=[LoggingInterceptor()]
)

Client-Side Interceptors

Clients also support interceptors for request/response processing:

client = HelloServiceClient(
    "https://api.example.com",
    interceptors=[AuthInterceptor(), RetryInterceptor()]
)

Advanced Features

Connect GET Support

Connect automatically enables GET request support for methods marked with idempotency_level = NO_SIDE_EFFECTS in your proto files:

service YourService {
  // This method will support both GET and POST requests
  rpc GetData(Request) returns (Response) {
    option idempotency_level = NO_SIDE_EFFECTS;
  }
}

Clients can use GET requests automatically:

# The client will use GET for idempotent methods
response = await client.get_data(request)

CORS Support

Connect works with any ASGI CORS middleware. For example, using Starlette:

from starlette.middleware.cors import CORSMiddleware
from starlette.applications import Starlette

app = Starlette()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)
# Mount your Connect application

Message Size Limits

Protect against resource exhaustion by limiting message sizes:

# ASGI application with 1MB limit
app = YourServiceASGIApplication(
    service,
    read_max_bytes=1024 * 1024  # 1MB
)

# Client with message size limit
client = YourServiceClient(
    "https://api.example.com",
    read_max_bytes=1024 * 1024
)

When exceeded, returns RESOURCE_EXHAUSTED error.

Proto Editions Support

protoc-gen-connectrpc supports up to Protobuf Editions 2024:

edition = "2024";

package your.service;

service YourService {
  rpc YourMethod(Request) returns (Response);
}

Development

We use ruff for linting and formatting, ty for type checking, and tombi for TOML linting and formatting.

We rely on the conformance test suit (in ./conformance) to verify behavior.

Set up a virtual env:

uv sync

Then, use uv run poe check to do development checks, or check out uv run poe for other targets.

Status

This project is in beta and is being actively developed. 1.0 will include a new Protobuf implementation built from scratch by Buf, which may introduce breaking changes. Join us on Slack if you have questions or feedback.

Legal

Offered under the Apache 2 license.