🚀

PydanticRPCでgRPC/Connect RPCサービスを簡単開発

2025/01/28に公開

はじめまして。Zennでの初めての記事投稿です。

今回、拙作の PydanticRPC というPythonライブラリをご紹介します。
このライブラリを使うと、Pydantic で定義したモデルの型情報をもとに.protoファイルとRPCサービス(gRPCやConnect RPC)を自動生成し、.protoファイルを手書きすることなくRPCサービス(API)をつくれます

https://github.com/i2y/pydantic-rpc

はじめに

REST APIをPythonで書こうと思ったときに、FastAPIやFlaskなどを利用することが多いと思います。
一方、データ通信をより効率的に行いたい場合やスキーマファースト開発したい場合などには gRPCConnect RPCなどを選択することもあると思います。

しかし、それらを利用するためには基本的に .proto ファイル(Protocol Buffers)を書く必要があります。

  • IDL(Interface Definition Language)である .proto ファイルを定義し
  • protocbuf でコード生成し
  • それをアプリケーションに組み込んで…

というフローは非常に強力である反面、手間や学習コストもそれなりにかかります

そこで登場するのが、今回ご紹介する PydanticRPC です。
.proto を手書かずとも、Pydanticモデルに基づいて実行時(サービス起動時)にProtobufファイルを生成 し、そのままサーバーを起動できます。

PydanticRPCとは?

PydanticRPC は以下の特徴を持ったPythonライブラリです。

  • 自動Protobuf生成
    RPCサービスとして公開/提供するPythonオブジェクトとPydanticモデルのシグネチャから .proto を自動生成
  • 動的コード生成
    内部的に grpcio-tools を用いてサーバー/クライアントのスタブを自動生成し、サーバースタブと公開対象Pythonオブジェクトを紐づけ
  • gRPCとConnect RPCに対応
    gRPC はもちろん、gRPC-WebConnect RPCConnecpy)でサーブ。非同期(asyncio)にも対応

つまり、「PythonでクラスやPydanticモデルを書いたら、そのままRPCサービスが作れてしまう」 という、便利な仕組みが手に入るわけです。

インストール

PyPIに公開されているため、pipコマンドなどでインストールできます。

pip install pydantic-rpc

使い方:gRPCサービスをつくる

pydantic_rpc.Server を使うことで、gRPCサーバーを立ち上げられます。

サーバーの例

# 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はPydanticモデルでバリデーション済み
        return HelloReply(message=f"Hello, {request.name}!")

if __name__ == "__main__":
    server = Server()
    # Greeterオブジェクトを登録してサーバー起動
    server.run(Greeter())
  • Messagepydantic.BaseModel のエイリアスで、BaseModelそのもの
  • Greeter は公開したいメソッドを持つクラス
    • say_hello メソッドは (HelloRequest) -> HelloReply のシグネチャを持つ
  • Server().run(Greeter()) を呼び出すと、PydanticRPCが内部的に .proto を生成し、gRPCサーバーを起動

サーバーが起動すると localhost:50051(デフォルト)で接続を待ち受けているので、任意のgRPCクライアントから呼び出せます。

非同期(asyncio)サーバーの例

非同期サーバーを書きたい場合は AsyncIOServer を利用します。
メソッド定義を async def にするだけでOKです。

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:
        # 非同期の何らかの処理を書ける
        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()) をコルーチンとして呼び出し、イベントループに渡すだけです。

使い方:Connect RPCサービスをつくる

Connecpy による Connect RPC に対応したASGIアプリケーションを構築することも可能です。

PydanticAIを用いて開催年から五輪開催地を回答するASGIアプリケーションの例です。 pydantic_rpc.ConnecpyASGIApp を使います。

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

# ASGIアプリとして公開
app = ConnecpyASGIApp()
app.mount(OlympicsLocationAgent())
  • ConnecpyASGIApp を生成し、Connect RPC でサービスとして公開したいクラスを mount(...)
  • クライアントからリクエストを送信すると、Pydanticモデルでバリデーション済みのデータが受け渡される

FastAPIやStarletteなどと組み合わせる場合は、生成したアプリ app をルーティングに組み込めばOKです。これにより、「一部のエンドポイントはConnect RPCで公開しつつ、他のエンドポイントはRESTで…」といった混在構成も、比較的簡単に実現できます。

使い方:gRPC-Webサービスをつくる

PydanticRPCは gRPC-Web サービスを WSGI, ASGI アプリケーションにマウントする仕組みも提供しています。

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):
    # ここにはFastAPIやStarletteなどの既存アプリを定義
    pass

# 既存アプリにアタッチ
app = ASGIApp(your_asgi_app)
app.mount(Greeter())

これにより、gRPC-Webアプリを既存のフレームワーク(Flask, Django, FastAPIなど)に組み込み可能です。こちらも Connect RPCでの場合と同様に、「一部のエンドポイントはgRPC-Webで公開しつつ、他のエンドポイントはRESTで…」といった混在構成を、比較的簡単に実現できます。

まとめ

拙作の PydanticRPC をご紹介しました。もし興味を持たれた方は、ぜひ PydanticRPCのGitHubリポジトリ をチェックしていただけると幸いです。

Discussion