gRPC入門｜Pythonで4通信方式を実装、RESTとの使い分けも

gRPCとは、「.proto ファイルにサービスの形を書く → 各言語のコードを自動生成する → HTTP/2 の上で Protocol Buffers 形式のバイナリをやり取りする」という3点セットで、サービス間の呼び出しを関数呼び出しのように書ける RPC フレームワークです。 REST のように URL とステータスコードを設計する代わりに、rpc GetTask (GetTaskRequest) returns (Task); と1行書けば、サーバーの土台とクライアントの呼び出し口の両方が生成されます。

この記事では、gRPC・Protocol Buffers・HTTP/2 の関係を図と表で整理したうえで、Python（grpcio 1.84.0、2026年10月時点の PyPI 最新）で unary / server streaming / client streaming / bidirectional の4方式をひとつのサーバーとクライアントに全部入りで実装し、実行結果をそのまま掲載します。エラー処理・deadline・metadata・Interceptor、Node からの呼び出し、Reflection によるサービス一覧の取得、REST / GraphQL との使い分け、そして protoc や生成ファイルでつまずいたときの実際のエラー文と直し方まで、1本で追えるようにしました。

gRPC 4通信方式の早見表

通信方式は .proto の stream の位置で決まり、Python 側の書き方は「イテレータを受け取るか」「yield で返すか」の組み合わせです。

gRPC とは：Protocol Buffers と HTTP/2 の関係

gRPC は「呼び出しの約束ごと（フレームワーク）」、Protocol Buffers は「データの形と直列化」、HTTP/2 は「運ぶ通信路」という役割分担です。 3つはセットで使われますが、担当する層が違います。

flowchart LR
    subgraph C["クライアント（Python / Node など）"]
        A["アプリのコード<br/>stub.GetTask(req)"] --> S["Stub<br/>（.proto から自動生成）"]
    end
    subgraph T["通信路"]
        S -- "Protocol Buffers で<br/>バイナリに直列化" --> H["HTTP/2 ストリーム<br/>HEADERS = metadata<br/>DATA = メッセージ<br/>trailer = ステータスコード"]
    end
    subgraph V["サーバー（Python）"]
        H --> D["Servicer 基底クラス<br/>（.proto から自動生成）"] --> I["自分で実装した<br/>GetTask(request, context)"]
    end

HTTP/2 を使うため、1リクエスト1レスポンスだけでなく、サーバーから複数件を流す・クライアントから複数件を送る・双方向に流す、の4方式が同じ仕組みで扱えます。 ステータスコードは HTTP のものとは別に gRPC 独自の OK / NOT_FOUND / DEADLINE_EXCEEDED などが定義され、HTTP/2 のトレーラー（末尾ヘッダー）で返ります。

なお .proto には syntax = "proto3" のほかに「Protobuf Editions（最新は edition = "2024"）」という書式もありますが、proto3 のファイルはそのまま使え、移行も必須ではないと公式ドキュメントにあるため、この記事は proto3 で書きます。

環境準備と .proto からのコード生成

protoc 本体を別途入れる必要はなく、grpcio-tools に同梱されています。

python3 -m venv .venv && source .venv/bin/activate
pip install grpcio grpcio-tools grpcio-reflection
pip list | grep -i -E "grpc|protobuf"
python -m grpc_tools.protoc --version

実行結果（Python 3.11.15、2026年10月1日時点）:

grpcio            1.84.0
grpcio-reflection 1.84.0
grpcio-tools      1.84.0
protobuf          7.36.2
libprotoc 35.1

.proto を書く（タスク管理サービスの例）

protos/task.proto に、メッセージ4種類と「4通信方式が1つずつ入った」サービスを定義します。

syntax = "proto3";

package task.v1;

// ---- メッセージ定義 ----
enum Priority {
  PRIORITY_UNSPECIFIED = 0;
  LOW = 1;
  NORMAL = 2;
  HIGH = 3;
}

message Task {
  int32 id = 1;
  string title = 2;
  Priority priority = 3;
  bool done = 4;
}

message GetTaskRequest {
  int32 id = 1;
}

message ListTasksRequest {
  Priority min_priority = 1;  // この優先度以上のタスクを返す
}

message UploadSummary {
  int32 count = 1;       // 受け取った件数
  int32 high_count = 2;  // うち HIGH の件数
}

message Progress {
  int32 task_id = 1;
  int32 percent = 2;
}

message Notice {
  int32 task_id = 1;
  string message = 2;
}

// ---- サービス定義（4つの通信方式）----
service TaskService {
  // 1. Unary: 1リクエスト → 1レスポンス
  rpc GetTask (GetTaskRequest) returns (Task);
  // 2. Server streaming: 1リクエスト → 複数レスポンス
  rpc ListTasks (ListTasksRequest) returns (stream Task);
  // 3. Client streaming: 複数リクエスト → 1レスポンス
  rpc UploadTasks (stream Task) returns (UploadSummary);
  // 4. Bidirectional streaming: 複数リクエスト ⇔ 複数レスポンス
  rpc ReportProgress (stream Progress) returns (stream Notice);
}

コードを生成する

python -m grpc_tools.protoc -I protos --python_out=. --pyi_out=. --grpc_python_out=. protos/task.proto
ls task_pb2*

task_pb2.py  task_pb2.pyi  task_pb2_grpc.py

-I（--proto_path）は「import の基準ディレクトリ」で、生成ファイルの import 文にも影響します（後述のつまずき参照）。

4通信方式をサーバーとクライアントで実装する

サーバー（server.py）

4メソッドに加えて、context.abort によるエラー返却、metadata の読み書き、ログ用 Interceptor、Reflection を1ファイルに入れています。 ポートはこの記事では 7621 を使います。

"""gRPC サーバー（4通信方式 + エラー処理 + metadata + Interceptor + Reflection）
実行: python server.py   （ポート 7621 で待ち受け）
"""
import logging
import time
from concurrent import futures

import grpc
from grpc_reflection.v1alpha import reflection

import task_pb2
import task_pb2_grpc

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")

# インメモリの「DB」
TASKS = {
    1: task_pb2.Task(id=1, title="仕様を確認する", priority=task_pb2.HIGH),
    2: task_pb2.Task(id=2, title="設計書を書く", priority=task_pb2.NORMAL),
    3: task_pb2.Task(id=3, title="レビュー依頼", priority=task_pb2.LOW, done=True),
}


class TaskService(task_pb2_grpc.TaskServiceServicer):
    # 1. Unary
    def GetTask(self, request, context):
        # クライアントが送った metadata を読む
        md = dict(context.invocation_metadata())
        logging.info("GetTask id=%s x-request-id=%s", request.id, md.get("x-request-id"))
        task = TASKS.get(request.id)
        if task is None:
            # ステータスコード + 詳細メッセージを付けて中断（以降は実行されない）
            context.abort(grpc.StatusCode.NOT_FOUND, f"task {request.id} not found")
        # トレーラー metadata をレスポンスに添える
        context.set_trailing_metadata((("x-served-by", "server-7621"),))
        return task

    # 2. Server streaming: yield で複数返す
    def ListTasks(self, request, context):
        for task in TASKS.values():
            if task.priority < request.min_priority:
                continue
            if not context.is_active():  # クライアントが切断/タイムアウトしたら止める
                logging.info("ListTasks: client gone, stop streaming")
                return
            time.sleep(0.2)  # 重い処理のつもり
            yield task

    # 3. Client streaming: request_iterator を回して1つ返す
    def UploadTasks(self, request_iterator, context):
        count = high = 0
        for task in request_iterator:
            TASKS[task.id] = task
            count += 1
            high += task.priority == task_pb2.HIGH
        return task_pb2.UploadSummary(count=count, high_count=high)

    # 4. Bidirectional streaming: 受け取りながら yield で返す
    def ReportProgress(self, request_iterator, context):
        for p in request_iterator:
            if not 0 <= p.percent <= 100:
                context.abort(grpc.StatusCode.INVALID_ARGUMENT, f"percent must be 0-100, got {p.percent}")
            msg = "completed" if p.percent == 100 else f"{p.percent}% done"
            yield task_pb2.Notice(task_id=p.task_id, message=msg)


class LoggingInterceptor(grpc.ServerInterceptor):
    """全 RPC の呼び出しをログに残す最小の Interceptor"""

    def intercept_service(self, continuation, handler_call_details):
        logging.info("RPC %s", handler_call_details.method)
        return continuation(handler_call_details)


def serve(port: int = 7621):
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        interceptors=[LoggingInterceptor()],
    )
    task_pb2_grpc.add_TaskServiceServicer_to_server(TaskService(), server)
    # Reflection（grpcurl や動的クライアントがサービス一覧を取得できるようにする）
    service_names = (
        task_pb2.DESCRIPTOR.services_by_name["TaskService"].full_name,
        reflection.SERVICE_NAME,
    )
    reflection.enable_server_reflection(service_names, server)
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    logging.info("listening on %s", port)
    server.wait_for_termination()


if __name__ == "__main__":
    serve()

クライアント（client.py）

4方式の呼び出しに加え、with_call によるトレーラー metadata の取得、RpcError の処理、timeout、接続先なしの UNAVAILABLE、Unary 呼び出しに metadata を自動付与する Interceptor まで通しで動かします。

"""gRPC クライアント（4通信方式 + エラー処理 + deadline + metadata + Interceptor）
実行: python client.py
"""
import time

import grpc

import task_pb2
import task_pb2_grpc


class RequestIdInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Unary 呼び出しに x-request-id を自動で付ける最小の Interceptor"""

    def intercept_unary_unary(self, continuation, client_call_details, request):
        md = list(client_call_details.metadata or [])
        md.append(("x-request-id", f"req-{int(time.time() * 1000)}"))
        details = client_call_details._replace(metadata=md)
        return continuation(details, request)


def main():
    channel = grpc.insecure_channel("127.0.0.1:7621")
    channel = grpc.intercept_channel(channel, RequestIdInterceptor())
    stub = task_pb2_grpc.TaskServiceStub(channel)

    # 1. Unary（with_call でレスポンスと call の両方を受け取る）
    print("--- 1. Unary: GetTask")
    task, call = stub.GetTask.with_call(task_pb2.GetTaskRequest(id=1))
    print(f"id={task.id} title={task.title} priority={task_pb2.Priority.Name(task.priority)}")
    print("trailing metadata:", dict(call.trailing_metadata()))

    # 2. Server streaming（戻り値はイテレータ）
    print("--- 2. Server streaming: ListTasks(min_priority=NORMAL)")
    for t in stub.ListTasks(task_pb2.ListTasksRequest(min_priority=task_pb2.NORMAL)):
        print(f"  <- id={t.id} title={t.title}")

    # 3. Client streaming（イテレータ/ジェネレータを渡す）
    print("--- 3. Client streaming: UploadTasks")
    new_tasks = [
        task_pb2.Task(id=10, title="テストを書く", priority=task_pb2.HIGH),
        task_pb2.Task(id=11, title="READMEを更新", priority=task_pb2.LOW),
    ]
    summary = stub.UploadTasks(iter(new_tasks))
    print(f"count={summary.count} high_count={summary.high_count}")

    # 4. Bidirectional streaming（送るジェネレータを渡し、返ってくるイテレータを回す）
    print("--- 4. Bidirectional: ReportProgress")

    def progress_gen():
        for pct in (30, 60, 100):
            yield task_pb2.Progress(task_id=10, percent=pct)
            time.sleep(0.1)

    for notice in stub.ReportProgress(progress_gen()):
        print(f"  <- task {notice.task_id}: {notice.message}")

    # 5. エラー処理: 存在しない id → NOT_FOUND
    print("--- 5. Error: GetTask(id=999)")
    try:
        stub.GetTask(task_pb2.GetTaskRequest(id=999))
    except grpc.RpcError as e:
        print(f"code={e.code()} details={e.details()!r}")

    # 6. Deadline: 0.3秒で打ち切る（サーバーは1件0.2秒かかる）
    print("--- 6. Deadline: ListTasks(timeout=0.3)")
    try:
        for t in stub.ListTasks(task_pb2.ListTasksRequest(), timeout=0.3):
            print(f"  <- id={t.id}")
    except grpc.RpcError as e:
        print(f"code={e.code()} details={e.details()!r}")

    # 7. 接続先が無い → UNAVAILABLE
    print("--- 7. UNAVAILABLE: 127.0.0.1:7622 (nothing listening)")
    dead = task_pb2_grpc.TaskServiceStub(grpc.insecure_channel("127.0.0.1:7622"))
    try:
        dead.GetTask(task_pb2.GetTaskRequest(id=1), timeout=2)
    except grpc.RpcError as e:
        print(f"code={e.code()} details={e.details()!r}")


if __name__ == "__main__":
    main()

実行結果

別ターミナルで python server.py を起動してから python client.py を実行した結果です（grpcio 1.84.0）。

--- 1. Unary: GetTask
id=1 title=仕様を確認する priority=HIGH
trailing metadata: {'x-served-by': 'server-7621'}
--- 2. Server streaming: ListTasks(min_priority=NORMAL)
  <- id=1 title=仕様を確認する
  <- id=2 title=設計書を書く
--- 3. Client streaming: UploadTasks
count=2 high_count=1
--- 4. Bidirectional: ReportProgress
  <- task 10: 30% done
  <- task 10: 60% done
  <- task 10: completed
--- 5. Error: GetTask(id=999)
code=StatusCode.NOT_FOUND details='task 999 not found'
--- 6. Deadline: ListTasks(timeout=0.3)
  <- id=1
code=StatusCode.DEADLINE_EXCEEDED details='Stream removed (Deadline Exceeded)'
--- 7. UNAVAILABLE: 127.0.0.1:7622 (nothing listening)
code=StatusCode.UNAVAILABLE details='failed to connect to all addresses; last error: UNKNOWN: ipv4:127.0.0.1:7622: Failed to connect to remote host: Connection refused'

サーバー側のログ（Interceptor が全 RPC を記録し、GetTask では Interceptor が付けた x-request-id が読めています）:

2026-10-01 07:05:15,255 listening on 7621
2026-10-01 07:05:17,226 RPC /task.v1.TaskService/GetTask
2026-10-01 07:05:17,227 GetTask id=1 x-request-id=req-1790805917224
2026-10-01 07:05:17,228 RPC /task.v1.TaskService/ListTasks
2026-10-01 07:05:17,634 RPC /task.v1.TaskService/UploadTasks
2026-10-01 07:05:17,639 RPC /task.v1.TaskService/ReportProgress
2026-10-01 07:05:17,947 RPC /task.v1.TaskService/GetTask
2026-10-01 07:05:17,947 GetTask id=999 x-request-id=req-1790805917946
2026-10-01 07:05:17,948 RPC /task.v1.TaskService/ListTasks

Server streaming の結果が「id=1 の1件だけ」で DEADLINE_EXCEEDED になっているのが deadline の効果です。 サーバーは1件 0.2 秒かかるので、0.3 秒の期限内に届いた1件だけを受け取り、2件目を待っている途中で打ち切られています。

エラー処理・deadline・metadata・Interceptor

ステータスコードの使い分け

gRPC のエラーは HTTP ステータスではなく grpc.StatusCode で表し、サーバーは context.abort(code, details)、クライアントは grpc.RpcError の code() と details() で受け取ります。 よく使うコードを場面別にまとめます。

deadline（タイムアウト）

gRPC のタイムアウトは「期限（deadline）」で、クライアントが timeout=秒 を指定するとサーバーにも伝わり、期限を過ぎた RPC は両側で打ち切られます。 実行結果のとおり、ストリーミングの途中でも DEADLINE_EXCEEDED になります。details() の文字列は同じ環境でも 'Deadline Exceeded' と 'Stream removed (Deadline Exceeded)' の両方が出ました（3回実行）。判定には必ず code() を使います。

metadata（ヘッダー相当）

metadata は HTTP ヘッダーに相当するキーと値の並びで、認証トークンや相関 ID を本文と分けて送るのに使います。 キーは小文字、値は文字列（バイナリは末尾 -bin のキーで bytes）です。

Interceptor（共通処理の差し込み）

Interceptor は全 RPC の前後に共通処理を差し込む仕組みで、ログ・認証・metadata の自動付与・再試行に使います。 サーバー側は grpc.ServerInterceptor.intercept_service で handler_call_details.method（/task.v1.TaskService/GetTask の形）が取れ、continuation(...) を呼ばずに別ハンドラを返せば認証失敗の応答も作れます。クライアント側は通信方式ごとに UnaryUnaryClientInterceptor など4種類があり、grpc.intercept_channel(channel, ...) で束ねます。上のコードは Unary だけが対象なので、ストリーミング呼び出しには x-request-id が付きません。

非同期（asyncio）で書く

grpc.aio を使うと await と async for で同じ Stub が使えます。 サーバーも grpc.aio.server() と async def で書けます（サーバー・クライアントで同期と非同期を混ぜても通信できます）。

import asyncio, grpc, task_pb2, task_pb2_grpc

async def main():
    async with grpc.aio.insecure_channel("127.0.0.1:7621") as ch:
        stub = task_pb2_grpc.TaskServiceStub(ch)
        task = await stub.GetTask(task_pb2.GetTaskRequest(id=2))
        print("unary:", task.title)
        async for t in stub.ListTasks(task_pb2.ListTasksRequest(min_priority=task_pb2.HIGH)):
            print("stream:", t.id, t.title)

asyncio.run(main())

unary: 設計書を書く
stream: 1 仕様を確認する
stream: 10 テストを書く

（id=10 は先ほど UploadTasks で追加した HIGH のタスクです。）

Node（@grpc/grpc-js）から同じサーバーを呼ぶ

gRPC の利点のひとつが「同じ .proto から別言語のクライアントが作れる」ことです。 Node では @grpc/grpc-js（1.14.5）と @grpc/proto-loader（0.8.1、いずれも 2026年10月時点の npm 最新）を使い、.proto を実行時に読み込む方式ならコード生成なしで呼べます。

npm install @grpc/grpc-js @grpc/proto-loader

// client.mjs — Node から Python の gRPC サーバーを呼ぶ
import grpc from "@grpc/grpc-js";
import protoLoader from "@grpc/proto-loader";

const def = protoLoader.loadSync("../protos/task.proto", {
  keepCase: true, longs: String, enums: String, defaults: true, oneofs: true,
});
const { task } = grpc.loadPackageDefinition(def);
const client = new task.v1.TaskService("127.0.0.1:7621", grpc.credentials.createInsecure());

// Unary はコールバック（Promise 化して使う）
const getTask = (id) => new Promise((resolve, reject) =>
  client.GetTask({ id }, (err, res) => (err ? reject(err) : resolve(res))));

console.log("GetTask(1):", await getTask(1));
try {
  await getTask(999);
} catch (e) {
  console.log(`GetTask(999): code=${e.code} (${grpc.status[e.code]}) details=${e.details}`);
}

// Server streaming はイベント
const stream = client.ListTasks({ min_priority: "NORMAL" });
stream.on("data", (t) => console.log("ListTasks <-", t.id, t.title));
stream.on("end", () => { console.log("ListTasks end"); client.close(); });

実行結果（Node 22.22.2。Python サーバーは起動したまま）:

GetTask(1): { id: 1, title: '仕様を確認する', priority: 'HIGH', done: false }
GetTask(999): code=5 (NOT_FOUND) details=task 999 not found
ListTasks <- 1 仕様を確認する
ListTasks <- 2 設計書を書く
ListTasks <- 10 テストを書く
ListTasks end

Python 側で context.abort したエラーが、Node 側でも同じコード 5（NOT_FOUND）と同じ details で受け取れています。 ステータスコードと metadata の仕様が言語をまたいで共通なのが gRPC の強みです。enums: String を指定しているので enum は 'HIGH' の文字列で返り、リクエストにも "NORMAL" と文字列で渡せます。

grpcurl の代わりに Reflection でサービス一覧を取得する

REST の curl にあたるツールが grpcurl です（grpcurl -plaintext 127.0.0.1:7621 list でサービス一覧、describe で定義、-d '{"id":1}' で呼び出し）。今回の検証環境では GitHub からバイナリを取得できなかったため grpcurl は未実行ですが、grpcurl が使う仕組みは「Server Reflection」という gRPC 標準のサービスなので、Python からも同じ情報が取れます。サーバー側では grpcio-reflection の enable_server_reflection を呼ぶだけです（server.py 参照）。

"""grpcurl の代わりに Python の Reflection でサービス一覧とメソッドを取得する"""
import grpc
from google.protobuf.descriptor_pool import DescriptorPool
from grpc_reflection.v1alpha.proto_reflection_descriptor_database import (
    ProtoReflectionDescriptorDatabase,
)

channel = grpc.insecure_channel("127.0.0.1:7621")
db = ProtoReflectionDescriptorDatabase(channel)
pool = DescriptorPool(db)

# grpcurl list に相当
for name in db.get_services():
    print("service:", name)

# grpcurl describe task.v1.TaskService に相当
svc = pool.FindServiceByName("task.v1.TaskService")
for m in svc.methods:
    kind = ("stream " if m.client_streaming else "") + m.input_type.name
    kind += " -> " + ("stream " if m.server_streaming else "") + m.output_type.name
    print(f"  rpc {m.name}({kind})")

service: grpc.reflection.v1alpha.ServerReflection
service: task.v1.TaskService
  rpc GetTask(GetTaskRequest -> Task)
  rpc ListTasks(ListTasksRequest -> stream Task)
  rpc UploadTasks(stream Task -> UploadSummary)
  rpc ReportProgress(stream Progress -> stream Notice)

.proto を持っていないクライアントでも、Reflection が有効ならサーバーから定義を取り出して呼び出せます。 開発・検証環境では有効にしておくと調査が楽になり、公開環境では無効にするか認証の内側に置く、という運用が一般的です。

REST・GraphQL・gRPC の使い分け

gRPC はサービス間の内部通信とストリーミングに強く、ブラウザ向けの公開 API は REST（または GraphQL）が扱いやすい、という住み分けが基本です。 どれかが上位互換ということはなく、境界ごとに選びます。

REST と GraphQL の詳しい比較は別記事で扱っているので、ここでは gRPC との境界に絞りました。

よくあるつまずきと対処（実際のエラー文）

つまずきの多くは「-I の指定」「生成ファイルの置き場所」「接続先」の3つに集まります。 実際に出したエラー文で確認します。

1. -I を付けずに protoc を実行した

python -m grpc_tools.protoc --python_out=. --grpc_python_out=. protos/task.proto

protos/task.proto: File does not reside within any path specified using --proto_path (or -I).  You must specify a --proto_path which encompasses this file.  Note that the proto_path must be an exact prefix of the .proto file names -- protoc is too dumb to figure out when two paths (e.g. absolute and relative) are equivalent (it's harder than you think).

→ .proto のパスは -I で指定したディレクトリからの相対パスでなければなりません。 -I protos ... protos/task.proto のように、-I の値が .proto パスの先頭と一致するように書きます。-I . ... task.proto のように一致しないと Could not make proto path relative: task.proto: No such file or directory になります。

2. 生成ファイルをサブディレクトリに置いたら import できない

--python_out=gen --grpc_python_out=gen で gen/ に生成し、from gen import task_pb2_grpc としたときのエラーです。

  File ".../gen/task_pb2_grpc.py", line 6, in <module>
    import task_pb2 as task__pb2
ModuleNotFoundError: No module named 'task_pb2'

→ 生成された task_pb2_grpc.py は -I を基準にした相対パスで task_pb2 を import します。 直し方は2つです。

3. StatusCode.UNAVAILABLE: failed to connect to all addresses

code=StatusCode.UNAVAILABLE details='failed to connect to all addresses; last error: UNKNOWN: ipv4:127.0.0.1:7622: Failed to connect to remote host: Connection refused'

→ サーバーが起動していない、ポートが違う、localhost が IPv6 に解決されてサーバーが IPv4 だけ待っている、TLS の有無が食い違っている（insecure_channel に対して add_secure_port）、のいずれかです。 サーバー側は [::]:7621 のように待ち受ければ IPv4/IPv6 の両方を受けられます。運用ではこのコードのときだけ再試行する設計にします。

4. 生成コードと grpcio の版が合わない

生成された task_pb2_grpc.py には版のチェックが入っており、生成に使った grpcio-tools より前の版の grpcio で実行すると次のメッセージで止まります（生成コード内のメッセージ）。

RuntimeError: The grpc package installed is at version X, but the generated code in task_pb2_grpc.py depends on grpcio>=1.84.0. Please upgrade your grpc module to grpcio>=1.84.0 or downgrade your generated code using grpcio-tools<=X.

→ grpcio と grpcio-tools は同じバージョンに揃え、生成ファイルはリポジトリに入れずビルド時に生成するか、生成に使った版を固定します。

protobuf ランタイム側にも同様のチェックがあり、生成コードより前のメジャー版のランタイム（例: protobuf 5.29.5 で今回の生成ファイルを import）では次のように止まります（実際のエラー文）。

google.protobuf.runtime_version.VersionError: Detected mismatched Protobuf Gencode/Runtime major versions when loading task.proto: gencode 7.35.1 runtime 5.29.5. Same major version is required. See Protobuf version guarantees at https://protobuf.dev/support/cross-version-runtime-guarantee.

よくある質問

Q. gRPC はブラウザから直接使えますか？ A. ブラウザの JavaScript は HTTP/2 のフレームやトレーラーを直接扱えないため、そのままでは使えません。gRPC-Web という変換方式と、それを gRPC に変換するプロキシ（Envoy など）を挟めば呼び出せます（Unary とサーバーストリーミングに対応）。ブラウザが主な利用者なら REST や GraphQL を選ぶのが素直です。

Q. protoc を別途インストールする必要はありますか？ A. Python では不要です。grpcio-tools に protoc 本体（今回の 1.84.0 では libprotoc 35.1）が同梱されていて、python -m grpc_tools.protoc で呼び出せます。Node の @grpc/proto-loader も .proto を実行時に読むので不要です。

Q. 通信の暗号化（TLS）はどう設定しますか？ A. サーバーは server.add_secure_port(addr, grpc.ssl_server_credentials([(秘密鍵, 証明書)]))、クライアントは grpc.secure_channel(addr, grpc.ssl_channel_credentials(root_certificates=CA証明書)) にします。自己署名証明書で試したところ TLS 同士は正常に通り、平文クライアントから TLS サーバーへ接続すると StatusCode.UNAVAILABLE（details は ... Socket closed）になりました。この記事の insecure_* は検証用です。

Q. .proto を変更したとき、以前のクライアントとの互換性はどう保ちますか？ A. フィールドを追加するときは新しい番号を使い、削除したフィールドの番号は reserved 4; で再利用を防ぎます。 型や番号を変えない限り、以前のクライアントは知らないフィールドを無視して動きます。メソッドの追加も互換ですが、削除・改名は UNIMPLEMENTED になるので、package task.v2 のように版を分けて移行します。

まとめ

• gRPC は「.proto で契約 → コード生成 → HTTP/2 + Protocol Buffers で通信」の3点セットで、Protocol Buffers が直列化、HTTP/2 が通信路、gRPC がその上のフレームワークです。
• 4通信方式は stream の位置で決まり、Python では「request_iterator を回す」「yield で返す」の組み合わせで書けます。今回のサーバー・クライアントは grpcio 1.84.0 で全方式の実行結果を確認しました。
• エラーは context.abort(StatusCode, details) で返し、grpc.RpcError の code() で判定します。details() の文字列は版や状況で変わるので判定に使いません。
• timeout は常に付ける。deadline はサーバーにも伝わり、ストリーミングの途中でも DEADLINE_EXCEEDED で打ち切られます。
• metadata と Interceptor で、認証・相関 ID・ログなどの共通処理を本文と分けて扱えます。
• 同じ .proto から Node のクライアントも作れ、ステータスコードと details は言語をまたいで同じでした。grpcurl が使えない環境でも grpcio-reflection で定義を取り出せます。
• 社内サービス間とストリーミングは gRPC、ブラウザや外部公開は REST / GraphQLが基本の使い分けで、つまずきは -I・生成先・接続先の3点を先に疑うと早く解けます。

株式会社RJCでは、gRPC のようなサービス間通信の技術を、実際に動かして結果を確かめながらプロジェクトの中で仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。