Skip to content

FastAPI

import uvicorn
from fastapi import FastAPI, Request, Response

from jsonrpcserver import Result, Success, async_dispatch, method

MAX_BODY = 1_000_000  # bytes

app = FastAPI()


@method
async def ping() -> Result:
    return Success("pong")


@app.post("/")
async def index(request: Request) -> Response:
    # FastAPI has no limit on the body size, so read it in chunks and stop
    # when it gets too big.
    body = b""
    async for chunk in request.stream():
        body += chunk
        if len(body) > MAX_BODY:
            return Response(status_code=413)
    # max_batch_size: see the Security page.
    if response := await async_dispatch(body.decode(), max_batch_size=100):
        return Response(response, media_type="application/json")
    return Response(status_code=204)


if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=8000)

FastAPI and Starlette have no limit on the body size, so the endpoint reads the body in chunks and answers 413 once it's too big. Many deployments also set a limit in the reverse proxy in front, such as nginx's client_max_body_size. max_batch_size limits how many requests one batch can hold. See Security.

The methods are async, and the endpoint uses async_dispatch. Plain methods work too, but they run on the event loop, so a slow one holds up other requests. See Async.

FastAPI's dependency injection doesn't reach into methods. Resolve what they need in the endpoint and pass it as context. Context has an example.

Try it

Save the example as fastapi_server.py, install FastAPI and Uvicorn (pip install fastapi uvicorn), and run it:

python fastapi_server.py

Then send it a request from another terminal:

curl -s -H 'Content-Type: application/json' -d '{"jsonrpc": "2.0", "method": "ping", "id": 1}' http://localhost:8000/
Output
{"jsonrpc": "2.0", "result": "pong", "id": 1}

The jsonrpcclient quickstart sends the same request from Python.