fastapi-openrpc

Declarative JSON-RPC 2.0 interface and OpenRPC 1.0 service discovery for FastAPI applications.

Features

• Declarative Registration — @method() decorator with automatic JSON Schema inference
• OpenRPC Service Discovery — GET /openrpc.json returns complete OpenRPC document
• Pydantic Support — Automatic schema generation with $ref references
• RpcContext Propagation — Secure caller context injection via FastAPI Depends()
• HMAC Authentication — Built-in signature verification with Nonce replay protection
• Role-Based Authorization — Method-level role declarations with composition semantics
• Business Errors — Structured error codes with errors array in OpenRPC document

Installation

pip install fastapi-openrpc

Optional dependencies:

# Redis support (Nonce replay protection)
pip install fastapi-openrpc[redis]

# Development dependencies
pip install fastapi-openrpc[dev]

Quick Start

from fastapi import FastAPI
from fastapi_openrpc import OpenRpcRouter, RpcContext, method

app = FastAPI()

openrpc_router = OpenRpcRouter(
    title="My API",
    version="1.0.0",
    description="My JSON-RPC service",
)

@method(name="getUser")
async def get_user(ctx: RpcContext, user_id: str) -> dict:
    """Get user information by user_id."""
    return {"id": user_id, "name": "Alice", "roles": ctx.roles}

@method(name="listUsers")
async def list_users(ctx: RpcContext, limit: int = 10) -> list[dict]:
    """Return a list of users."""
    return [{"id": i, "name": f"User {i}"} for i in range(limit)]

app.include_router(openrpc_router)

After starting the server, access GET /openrpc.json to view the OpenRPC service document.

Authentication

HMAC Authentication

from fastapi_openrpc.authentication import HMACAuthentication

auth = HMACAuthentication(secret_key="your-secret-key")
openrpc_router = OpenRpcRouter(
    authentication=auth,
)

Nonce Replay Protection

auth = HMACAuthentication(
    secret_key="your-secret-key",
    enable_nonce=True,
    redis_url="redis://localhost:6379",
    nonce_ttl=3600,
    connect_timeout=2.0,
    operation_timeout=5.0,
    cleanup_timeout=2.0,
)

Header format:

{
  "roles": ["admin"],
  "nonce": "550e8400-e29b-41d4-a716-446655440000"
}

Authorization

from fastapi_openrpc import Require

@method(name="adminAction", required_roles=["admin"])
async def admin_action(ctx: RpcContext) -> dict:
    """Admin only."""
    return {"status": "done"}

@method(name="editContent", required_roles=["editor", "admin"], require=Require.ALL)
async def edit_content(ctx: RpcContext) -> dict:
    """Requires both editor and admin roles."""
    return {"status": "edited"}

Error Handling

Built-in Errors

from fastapi_openrpc import InvalidParams, MethodNotFound

@method(name="divide")
async def divide(ctx: RpcContext, a: float, b: float) -> float:
    if b == 0:
        raise InvalidParams("Division by zero")
    return a / b

Built-in error codes:

Error	Code
Parse error	-32700
Invalid Request	-32600
Method Not Found	-32601
Invalid Params	-32602
Internal Error	-32603
Forbidden	-32000
Replay Detected	-32002

Business Errors

Define custom errors by subclassing BusinessError:

from fastapi_openrpc import BusinessError

# Define error codes starting from -32100
class UserNotFoundError(BusinessError):
    code = -32101
    message = "User not found"

    def __init__(self, user_id: str):
        super().__init__(data={"user_id": user_id})

Register error classes with methods using the errors parameter:

from examples.errors import UserNotFoundError  # Your error class

@method(name="getUser", errors=[UserNotFoundError])
async def get_user(ctx: RpcContext, user_id: str) -> dict:
    user = db.get(user_id)
    if not user:
        raise UserNotFoundError(user_id=user_id)
    return user

The errors array appears in the OpenRPC document for that method, enabling SDK auto-generation of typed exceptions:

{
  "name": "getUser",
  "errors": [
    {"code": -32101, "message": "User not found"}
  ]
}

Library reserves -32000 ~ -32099, application custom errors start from -32100.

API Reference

OpenRpcRouter

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)
openrpc_router = OpenRpcRouter(
    title="My API",
    version="1.0.0",
    description="...",
    servers=[{"url": "https://api.example.com"}],
    authentication=HMACAuthentication(secret_key="..."),  # Optional
    executor=executor,  # Optional, for sync handler thread pool
)
app.include_router(openrpc_router)

@method() Decorator

@method(
    name="myMethod",           # RPC method name
    tags=["admin"],           # OpenRPC tags
    summary="Short description",
    deprecated=False,
    required_roles=["admin"],  # Required roles
    require=Require.ALL,      # Role composition semantics
    errors=[MyError],        # Business error subclasses this method may raise
)
async def my_handler(ctx: RpcContext, arg1: str, arg2: int = 10) -> dict:
    return {"result": arg1}

Note: ctx: RpcContext is a reserved parameter name for injecting caller context.

Usage Examples

# Call RPC method
curl -X POST http://localhost:8000/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "getUser", "params": {"user_id": "42"}}'

# Service discovery
curl http://localhost:8000/openrpc.json

Documentation

For detailed user documentation, see docs/index.md.

Additional guides:

• Tutorial: Pydantic Models — 4 scenarios covering flat params, nested models, optional fields, and validation
• How-to: Error Handling — custom errors, Nonce replay protection, best practices
• Explanation: Concurrency Model — ThreadPoolExecutor, sync handler wrapping, closure safety

Contributing

Issues and Pull Requests are welcome.

License

MIT