中文 | English

tRPC-FastAPI

tRPC-FastAPI is a FastAPI integration plugin for the tRPC-Python framework (trpc_sdk). It provides HTTP serving out of the box, so you can develop, debug, deploy, and call HTTP APIs quickly. FastAPI is used as the server, so this plugin supports the server side only. Any HTTP client can call it.

Features

• FastAPI integration:
  • Drop-in fit: Integrates with the tRPC-Python framework and keeps every tRPC feature;
  • HTTP routing: Supports every HTTP method (GET, POST, PUT, DELETE, HEAD, OPTIONS, and others);
  • Generated docs: Generates OpenAPI/Swagger docs, including interactive API testing;
  • Type safety: Full type-annotation support for a better development experience;
• Development tools:
  • Static files: Serves static assets such as HTML, CSS, JS, and images;
  • Templates: Integrates the Jinja2 template engine for dynamic page rendering;
  • Version info: Reads and displays the FastAPI and plugin versions;
• Deployment:
  • Containers: Supports Docker;
  • Cloud native: Supports Kubernetes and other cloud-native environments;

Installation

Requirements

• Python 3.8+
• Linux / macOS (tRPC-Python does not support Windows)
• tRPC-Python-SDK

Install

tRPC-Python is currently pinned as a git dependency to the commit referenced above, and pip installs it together with the plugin:

pip install -e .
# To install the extra that includes static templates:
# pip install -e ".[static]"

Quick start

Basic usage

This assumes you already have a generated tRPC project. For the full walkthrough and the one-command run, see examples/README.md. The main files are:

examples/
├── normal_http/
├── http_sse_service.py
├── http_websocket_service.py
├── trpc_main.py
├── run_examples.sh
├── trpc_python.yaml                 # one process, one uvicorn
├── trpc_python_merge_threads.yaml   # one process, multiple handler threads
├── trpc_python_separate_thread.yaml # one process, I/O and handlers split
└── trpc_python_separate.yaml        # multi-process separate

1. Register HTTP routes

You can register routes in a separate file. The example below is http_service.py:

from trpc_fastapi import fastapi_route

@fastapi_route("/fastapi_http", ["GET", "HEAD"], route_params={'response_model': dict})
async def fastapi_http(request: Request) -> dict:
    """Test fastapi_http http"""
    logger.debug(f"=== normal_http, http_req header: {request.headers} ===")
    ctx = get_trpc_ctx()
    msg = ctx.get_client_message()
    register_service_name = "trpc.trpc_python_group.helloworld.Greeter1"
    msg_info = f"\n\n test_route, register service_name: {register_service_name}"
    msg_info = f"{msg_info}, callee_service_name: {msg.callee_service_name}\n\n"
    logger.debug(f"=== normal_http, msg_info: {msg_info} ===")
    return {"message": "This fastapi_http for [GET, HEAD]"}

2. RPC-style HTTP service

Code generated by the stub tool includes protobuf objects. Point it at http_fastapi and it is reachable as-is. Calling through the stub looks like this. Full source: test_service_rpc:

from trpc_sdk.abc import TrpcContext
from trpc_sdk.client import Options
from trpc_sdk.serialization import SerializationType

async def test_sync_Greeter_SayHello():
    """Test Greeter of SayHello"""
    proxy = rpc.GreeterClientProxyImpl()
    req = pb.HelloRequest()
    req.msg = "[sync] Hi SayHello"
    options = (Options().set_target('ip://127.0.0.1:8000').set_protocol('http').set_serialization_type(
        SerializationType.PB).set_network('tcp').set_short_conn(True))
    ctx = TrpcContext()
    try:
        ret = await proxy.asyncSayHello(ctx, req, options)
        print(f"SayHello ret is {ret}")
        assert ret is not None
    except Exception as exc:  # pylint: disable=broad-exception-caught
        print(f"error {exc}")
        assert 0

curl:

curl -X POST http://127.0.0.1:8000/trpc.trpc_python_group.helloworld.Greeter/SayHello \
     -H 'Content-Type: application/json' -d '{"msg":"data"}'

Import the route-registration module in trpc_main.py. The import must happen before create_server:

# Importing registers the plugin, so this has to come first.
import trpc_fastapi as _
# Import the registered HTTP service.
import normal_http as _

To serve another protocol name through FastAPI (for example the a2a plugin), call register_protocol_to_fastapi at import time. This also has to happen before create_server:

from trpc_fastapi import register_protocol_to_fastapi

register_protocol_to_fastapi("a2a")

3. Configuration

See trpc_python.yaml. When global.processmodel is omitted, the server is a single process with one event loop, and uvicorn runs on one thread of the service process.

server:
  service:
    - name: trpc.trpc_python_group.helloworld.FastApiGreeter
      ip: 0.0.0.0
      port: 8000
      network: tcp
      protocol: http_fastapi

Process models

Config	global.processmodel	uvicorn
trpc_python.yaml	omitted	one for the whole service
trpc_python_merge_threads.yaml	single_process, handle_num: 2	one, started on serving thread 0
trpc_python_separate_thread.yaml	single_process, io_handle_type: separate	one, started on serving thread 0
trpc_python_separate.yaml	multi_process, io_handle_type: separate, handle_num: 2	one per rpc task, listening on 8000 with SO_REUSEPORT. The I/O process listens only on the tRPC port

Multi-process separate snippet:

global:
  processmodel:
    mode: multi_process
    io_handle_type: separate
    handle_num: 2

4. Start the server

In trpc_main.py, import the module that registers routes, such as http_service.py above.

cd examples
./run_examples.sh
# or start it by hand
python3 trpc_main.py --conf trpc_python.yaml
python3 trpc_main.py --conf trpc_python_separate.yaml

5. Call the service

• In a browser: http://{IP}:{Port}/docs (IP and Port come from the config file. If the address is 0.0.0.0, use the IP of your environment.)

• With curl:

  curl -X GET http://127.0.0.1:8000/details -v
  

Advanced features

SSE

• Server code: http_sse_service

  import asyncio
  from fastapi.responses import StreamingResponse
  
  from trpc_fastapi import fastapi_route
  
  async def event_generator():
      """
      Simulate async events and push one message every second.
      """
      counter = 0
      while True:
          counter += 1
          # SSE format: data: <message>\n\n
          yield f"data: 这是第 {counter} 条消息\n\n"
          await asyncio.sleep(1)
  
  @fastapi_route('/sse', ["GET"])
  async def sse_endpoint():
      """
      SSE endpoint. Returns the event stream.
      """
      return StreamingResponse(event_generator(), media_type='text/event-stream')
  

• Import it in trpc_main.py:

  import http_sse_service as _
  

• Client check:

  curl -X GET http://127.0.0.1:8000/sse -v
  

WebSocket

• Server code: http_websocket_service

  from fastapi import WebSocket
  from fastapi.responses import HTMLResponse
  
  from trpc_sdk.log import logger
  from trpc_fastapi import fastapi_route
  
  @fastapi_route("/ws", method="WEBSOCKET")
  async def websocket_endpoint(websocket: WebSocket):
      await websocket.accept()
      try:
          while True:
              data = await websocket.receive_text()
              await websocket.send_text(f"Message text was: {data}")
      except Exception as e:  # pylint: disable=broad-except
          logger.error(f"websocket error: {e}")
  

• Import it in trpc_main.py:

  import http_websocket_service as _
  

• Client check:

  python3 examples/client/test_websocket_client.py
  

Static files

See static_http_service.

• Install

  Static pages may need template rendering. This install is the one to use:

  pip install -e ".[static]"
  

• Register the static directory

  from trpc_fastapi import fastapi_register_static
  # Directory of static files.
  static_dir = Path(__file__).parent / "static"
  # Files used by template rendering.
  templates_dir = Path(__file__).parent / "templates"
  
  fastapi_register_static("/static", static_dir)
  

• Register the static route

  from trpc_fastapi import fastapi_register_get, trpc_fastapi_version, get_fastapi_app
  
  @fastapi_register_get("/", route_params={'response_class': HTMLResponse})
  async def static_page(request: Request):
      """Root path. Returns an HTML page."""
      return templates.TemplateResponse(
          request,
          "static_page.html", {
              "trpc_fastapi_version": trpc_fastapi_version,
              "fastapi_version": fastapi_version,
              "endpoints": get_all_endpoints()
          })
  

  Call get_fastapi_app() inside an async def route. FastAPI runs synchronous routes in a thread pool, where the app bound to the current uvicorn thread is not available.

• In a browser: http://{IP}:{Port}/ (IP and Port come from the config file. If the address is 0.0.0.0, use the IP of your environment.)

Development setup

# After cloning, enter the project directory and install dev dependencies.
pip install -e ".[all]"

# Run tests
pytest tests

# Format the code
./format.sh

Commit conventions

• feat: for a new feature
• fix: for a bug fix
• docs: for a documentation update

License

This project is licensed under the Apache-2.0 License. See the LICENSE file.