fastapi-openrpc 0.1.4

FastAPI JSON-RPC 2.0 + OpenRPC 1.0 service discovery plugin

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

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,
    redis_required=True,       # Reject requests when Redis is unavailable
    connect_timeout=2.0,      # Redis connection timeout (seconds)
    operation_timeout=5.0,     # Redis operation timeout (seconds)
    cleanup_timeout=2.0,      # Redis cleanup timeout (seconds)
)

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

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

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
)
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