Hello DEV Community! Introducing PydanticRPC: Build gRPC & Connect RPC Services Without Manually Writing Protobuf Files

Tis is my first post on DEV, and I'd like to introduce you to one of my projects: PydanticRPC. This Python library automatically generates gRPC or Connect RPC services from Pydantic models—no need to handcraft .proto files!

GitHub – PydanticRPC

Overview

When creating a REST API in Python, you typically reach for frameworks like FastAPI or Flask. However, in scenarios where you want more efficient data communication or a schema-first approach, you might opt for gRPC or Connect RPC.

The usual workflow for using these RPC systems involves:

• Defining .proto files (Protocol Buffers)
• Generating code via protoc or buf
• Integrating that code into your application

While powerful, this process can be labor-intensive and comes with a learning curve.

That’s where PydanticRPC comes in. Without manually writing .proto, you can define your RPC data structures using Pydantic models, and PydanticRPC will generate the corresponding Protobuf definitions at runtime—then immediately spin up the server.

What Is PydanticRPC?

PydanticRPC is a Python library with the following key features:

• Automatic Protobuf Generation Creates .proto files on the fly from the signatures of your Python classes and Pydantic models that you want to expose as RPC services.
• Dynamic Code Generation Internally uses grpcio-tools to generate server/client stubs, then wires your Python classes to those stubs automatically.
• Supports gRPC, gRPC-Web, Connect RPC, and Async Works with standard gRPC, gRPC-Web (powered by Sonora, and Connect RPC (powered by Connecpy. It also supports asynchronous (asyncio) usage, including server-streaming methods.

In other words, “Write a Python class with Pydantic models, and you instantly get an RPC service without ever touching a .proto file!”

Installation

PydanticRPC is available on PyPI:

pip install pydantic-rpc

Usage: Creating a gRPC Service

You can stand up a gRPC server with pydantic_rpc.Server.

Synchronous Example

# server.py
from pydantic_rpc import Server, Message

class HelloRequest(Message):
    name: str

class HelloReply(Message):
    message: str

class Greeter:
    def say_hello(self, request: HelloRequest) -> HelloReply:
        # request is already validated by Pydantic
        return HelloReply(message=f"Hello, {request.name}!")

if __name__ == "__main__":
    server = Server()
    # Register your Greeter object and start the server
    server.run(Greeter())

• Message is an alias for pydantic.BaseModel (essentially the same).
• Greeter is the class whose methods you want to expose.
  • The say_hello method must follow the signature (HelloRequest) -> HelloReply.
• Calling Server().run(Greeter()) automatically generates the .proto file at runtime and starts a gRPC server on localhost:50051 by default.

Asynchronous Example

If you’d like an async server, use AsyncIOServer. Simply define your methods with async def:

import asyncio
from pydantic_rpc import AsyncIOServer, Message

class HelloRequest(Message):
    name: str

class HelloReply(Message):
    message: str

class Greeter:
    async def say_hello(self, request: HelloRequest) -> HelloReply:
        # Perform any async operations here
        return HelloReply(message=f"Hello, {request.name}!")

if __name__ == "__main__":
    server = AsyncIOServer()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(server.run(Greeter()))

server.run(Greeter()) is a coroutine, so you just run it in your event loop.

Usage: Response Streaming with PydanticRPC

PydanticRPC can also handle server-streaming responses (currently only supported in async gRPC), allowing your service to send messages to the client incrementally.

Below is an example using pydantic_ai to answer questions about the Olympics. It demonstrates both a standard method (ask) and a stream responses method (ask_stream), which returns an AsyncIterator of results:

import asyncio
from typing import AsyncIterator

from pydantic import field_validator
from pydantic_ai import Agent
from pydantic_rpc import AsyncIOServer, Message

# `Message` is just a pydantic BaseModel alias
class CityLocation(Message):
    city: str
    country: str

class OlympicsQuery(Message):
    year: int

    def prompt(self):
        return f"Where were the Olympics held in {self.year}?"

    @field_validator("year")
    def validate_year(cls, value):
        if value < 1896:
            raise ValueError("The first modern Olympics was held in 1896.")
        return value

class OlympicsDurationQuery(Message):
    start: int
    end: int

    def prompt(self):
        return f"From {self.start} to {self.end}, how many Olympics were held? Please provide the list of countries and cities."

    @field_validator("start")
    def validate_start(cls, value):
        if value < 1896:
            raise ValueError("The first modern Olympics was held in 1896.")
        return value

    @field_validator("end")
    def validate_end(cls, value):
        if value < 1896:
            raise ValueError("The first modern Olympics was held in 1896.")
        return value

class StreamingResult(Message):
    answer: str

class OlympicsAgent:
    def __init__(self):
        self._agent = Agent("ollama:llama3.2")

    async def ask(self, req: OlympicsQuery) -> CityLocation:
        # Standard unary method
        result = await self._agent.run(req.prompt(), result_type=CityLocation)
        return result.data

    async def ask_stream(
        self, req: OlympicsDurationQuery
    ) -> AsyncIterator[StreamingResult]:
        # Streaming method that yields partial results
        async with self._agent.run_stream(req.prompt(), result_type=str) as result:
            async for data in result.stream_text(delta=True):
                yield StreamingResult(answer=data)

if __name__ == "__main__":
    s = AsyncIOServer()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(s.run(OlympicsAgent()))

• ask(req: OlympicsQuery) -> CityLocation A normal unary RPC: the client sends a single year and gets back a single location.
• ask_stream(req: OlympicsDurationQuery) -> AsyncIterator[StreamingResult] A server-streaming RPC: the client sends a range of years, and the server yields multiple results over time.

When you run this server, PydanticRPC automatically generates a .proto file that defines both unary and response streaming methods, and it starts an async gRPC server on localhost:50051. Your client can then consume the streaming responses as they arrive.

Usage: Creating a Connect RPC Service

PydanticRPC also integrates with Connecpy to support Connect RPC in an ASGI app. Below is a simpler example:

from pydantic_ai import Agent
from pydantic_rpc import ConnecpyASGIApp, Message

class CityLocation(Message):
    city: str
    country: str

class Olympics(Message):
    year: int

    def prompt(self):
        return f"Where were the Olympics held in {self.year}?"

class OlympicsLocationAgent:
    def __init__(self):
        self._agent = Agent("ollama:llama3.2", result_type=CityLocation)

    async def ask(self, req: Olympics) -> CityLocation:
        result = await self._agent.run(req.prompt())
        return result.data

# Create an ASGI application for Connect RPC
app = ConnecpyASGIApp()
app.mount(OlympicsLocationAgent())

• Instantiate ConnecpyASGIApp(), then mount the class you want to expose as a Connect RPC service.
• Validation is handled by Pydantic, so your service methods receive validated data.
• If you already have an ASGI framework (FastAPI, Starlette, etc.), you can integrate this app by including it in your existing routing.

Usage: Creating a gRPC-Web Service

PydanticRPC can also serve gRPC-Web within WSGI or ASGI applications:

from pydantic_rpc import ASGIApp, Message

class HelloRequest(Message):
    name: str

class HelloReply(Message):
    message: str

class Greeter:
    def say_hello(self, req: HelloRequest) -> HelloReply:
        return HelloReply(message=f"Hello, {req.name}!")

async def your_asgi_app(scope, receive, send):
    # Some existing ASGI application (e.g., FastAPI, Starlette, etc.)
    pass

# Attach gRPC-Web to your existing app
app = ASGIApp(your_asgi_app)
app.mount(Greeter())

Now, gRPC-Web service endpoints and your existing REST endpoints can coexist. Similar to the Connect RPC approach, you can easily mix and match.

Conclusion

That’s an introduction to PydanticRPC—my Python library that enables gRPC, gRPC-Web, and Connect RPC services directly from Pydantic models. You can even implement server-streaming methods simply by using Python async generators. Future plans include further enhancements and more streaming features.

If this interests you, please check out the PydanticRPC GitHub repository. Thanks for reading, and feel free to reach out with any questions or feedback!

Source: View source