App

App

App(
    title: str = "Oxbrook",
    version: str = "0.1.0",
    description: str = "",
    openapi_url: str | None = "/openapi.json",
    docs_url: str | None = "/docs",
    mcp_url: str | None = "/mcp",
    debug: bool = False,
    access_log: bool = False,
    redis_url: str | None = None,
    lifespan: Any = None,
    worker_lifespan: Any = None,
    cors: CORS | None = None,
    websocket_origins: Any = None,
)

openapi_url and docs_url can each be set to None to disable them.

debug returns handler exception text in the 500 response. Leave it off outside development: exception messages routinely carry connection strings, file paths and user data.

access_log adds one log line per request. Opt-in: it costs something per request, and many deployments already log at the proxy.

redis_url enables durable topics. Nothing connects until the first durable topic is used.

mcp_url is where agents reach the service over the Model Context Protocol. It exposes only routes marked tool=True, plus topics as readable resources. Set it to None to turn the endpoint off entirely.

lifespan runs once around the server's life and worker_lifespan runs on every worker loop; what they yield is request.state. Two, because a connection pool belongs to the loop that made it and a server has several loops. See the lifespan guide.

cors lets pages on other origins call the app from a browser. Applied in Rust, to every response including the ones no handler produced.

websocket_origins lists the other origins whose pages may open a WebSocket. Browsers do not apply CORS to sockets and send cookies with the handshake, so without a check any website could open an authenticated socket as the user. A socket is accepted with no Origin (not a browser), from this server's own origin, or from a listed one; anything else is refused with 403 before an authorizer or handler runs. Left unset, the list is the CORS origins, excluding *. ["*"] turns the check off.

Source code in python/oxbrook/_app.py

def __init__(
    self,
    title: str = "Oxbrook",
    version: str = "0.1.0",
    description: str = "",
    openapi_url: str | None = "/openapi.json",
    docs_url: str | None = "/docs",
    mcp_url: str | None = "/mcp",
    debug: bool = False,
    access_log: bool = False,
    redis_url: str | None = None,
    lifespan: Any = None,
    worker_lifespan: Any = None,
    cors: CORS | None = None,
    websocket_origins: Any = None,
) -> None:
    """`openapi_url` and `docs_url` can each be set to None to disable them.

    `debug` returns handler exception text in the 500 response. Leave it off
    outside development: exception messages routinely carry connection
    strings, file paths and user data.

    `access_log` adds one log line per request. Opt-in: it costs something
    per request, and many deployments already log at the proxy.

    `redis_url` enables durable topics. Nothing connects until the first
    durable topic is used.

    `mcp_url` is where agents reach the service over the Model Context
    Protocol. It exposes only routes marked `tool=True`, plus topics as
    readable resources. Set it to None to turn the endpoint off entirely.

    `lifespan` runs once around the server's life and `worker_lifespan` runs
    on every worker loop; what they yield is `request.state`. Two, because
    a connection pool belongs to the loop that made it and a server has
    several loops. See the lifespan guide.

    `cors` lets pages on other origins call the app from a browser. Applied
    in Rust, to every response including the ones no handler produced.

    `websocket_origins` lists the other origins whose pages may open a
    WebSocket. Browsers do not apply CORS to sockets and send cookies with
    the handshake, so without a check any website could open an
    authenticated socket as the user. A socket is accepted with no
    `Origin` (not a browser), from this server's own origin, or from a
    listed one; anything else is refused with `403` before an authorizer or
    handler runs. Left unset, the list is the CORS origins, excluding `*`.
    `["*"]` turns the check off.
    """
    self.routes: list[RouteInfo] = []
    self._middleware: list[Any] = []
    self._exception_handlers: dict[type, Any] = {}
    if access_log:
        from ._logging import access_middleware

        # First registered, so it wraps everything and sees the final status.
        self._middleware.append(access_middleware)
    self._topics: dict[str, Topic] = {}
    self.title = title
    self.version = version
    self.description = description
    self.openapi_url = openapi_url
    self.docs_url = docs_url
    self.mcp_url = mcp_url
    self.debug = debug
    self.redis_url = redis_url
    self._backend: Any = None
    if cors is not None and not isinstance(cors, CORS):
        raise TypeError(f"cors must be a CORS(...), got {type(cors).__name__}")
    self.cors = cors
    if websocket_origins is not None:
        if isinstance(websocket_origins, str):
            raise TypeError("websocket_origins is a list of origins, not a single string")
        websocket_origins = tuple(check_origin(o) for o in websocket_origins)
    self.websocket_origins = websocket_origins
    self.lifespan = check_hook(lifespan, "lifespan")
    self.worker_lifespan = check_hook(worker_lifespan, "worker_lifespan")
    #: What `lifespan` yielded, while a server is running. Empty otherwise.
    self.state = State()

routes instance-attribute

routes: list[RouteInfo] = []

title instance-attribute

title = title

version instance-attribute

version = version

description instance-attribute

description = description

openapi_url instance-attribute

openapi_url = openapi_url

docs_url instance-attribute

docs_url = docs_url

mcp_url instance-attribute

mcp_url = mcp_url

debug instance-attribute

debug = debug

redis_url instance-attribute

redis_url = redis_url

cors instance-attribute

cors = cors

websocket_origins instance-attribute

websocket_origins = websocket_origins

lifespan instance-attribute

lifespan = check_hook(lifespan, 'lifespan')

worker_lifespan instance-attribute

worker_lifespan = check_hook(
    worker_lifespan, "worker_lifespan"
)

state instance-attribute

state = State()

topics property

topics: dict[str, Topic]

route

route(method: str, path: str, tool: bool = False)

Register a route.

tool=True also exposes it to agents over MCP. Opt-in on purpose: every route being agent-callable by default would mean an administrative delete endpoint is agent-callable by default.

Source code in python/oxbrook/_app.py

def route(self, method: str, path: str, tool: bool = False):
    """Register a route.

    `tool=True` also exposes it to agents over MCP. Opt-in on purpose:
    every route being agent-callable by default would mean an
    administrative delete endpoint is agent-callable by default.
    """
    method = method.upper()

    def decorator(fn):
        # Validates the handler against its path and fails here, at import
        # time, rather than on the first request.
        self._add(build_route(fn, method, path, tool=tool))
        return fn

    return decorator

get

get(path: str, tool: bool = False)

Source code in python/oxbrook/_app.py

def get(self, path: str, tool: bool = False):
    return self.route("GET", path, tool=tool)

post

post(path: str, tool: bool = False)

Source code in python/oxbrook/_app.py

def post(self, path: str, tool: bool = False):
    return self.route("POST", path, tool=tool)

put

put(path: str, tool: bool = False)

Source code in python/oxbrook/_app.py

def put(self, path: str, tool: bool = False):
    return self.route("PUT", path, tool=tool)

patch

patch(path: str, tool: bool = False)

Source code in python/oxbrook/_app.py

def patch(self, path: str, tool: bool = False):
    return self.route("PATCH", path, tool=tool)

delete

delete(path: str, tool: bool = False)

Source code in python/oxbrook/_app.py

def delete(self, path: str, tool: bool = False):
    return self.route("DELETE", path, tool=tool)

include

include(router: Router, prefix: str = '') -> None

Mount a router's routes, under prefix if given.

app.include(users.router, prefix="/api/v1")

Every route is validated against its full path here, and checked against the routes already registered, so a conflict between two modules fails at this call rather than at startup. Include after the router is fully declared: it cannot change afterwards.

Source code in python/oxbrook/_app.py

def include(self, router: Router, prefix: str = "") -> None:
    """Mount a router's routes, under `prefix` if given.

        app.include(users.router, prefix="/api/v1")

    Every route is validated against its full path here, and checked
    against the routes already registered, so a conflict between two
    modules fails at this call rather than at startup. Include after the
    router is fully declared: it cannot change afterwards.
    """
    if not isinstance(router, Router):
        raise TypeError(f"include() needs a Router, got {type(router).__name__}")
    for route in router._flatten(check_prefix(prefix), []):
        self._add(route)

middleware

middleware(fn)

Register middleware, which runs around every HTTP handler.

@app.middleware
async def require_key(request, call_next):
    if request.header("x-api-key") != SECRET:
        return Reply({"error": "unauthorized"}, status=401)
    return await call_next(request)

Runs outermost-first in registration order. WebSocket routes are not wrapped: their handshake completes before the handler runs, so there is nothing useful to intercept yet.

Source code in python/oxbrook/_app.py

def middleware(self, fn):
    """Register middleware, which runs around every HTTP handler.

        @app.middleware
        async def require_key(request, call_next):
            if request.header("x-api-key") != SECRET:
                return Reply({"error": "unauthorized"}, status=401)
            return await call_next(request)

    Runs outermost-first in registration order. WebSocket routes are not
    wrapped: their handshake completes before the handler runs, so there is
    nothing useful to intercept yet.
    """
    self._middleware.append(fn)
    return fn

exception_handler

exception_handler(exc_class: type)

Register how an exception becomes a response.

@app.exception_handler(LookupError)
async def missing(request, exc):
    return Reply({"error": "not found"}, status=404)

Applies to the class and its subclasses; the most specific registered class wins. Raised by a handler, a dependency, body validation, an authorizer or middleware, the exception is mapped before middleware sees the result, so the access log records the real status.

Registering HTTPError or RequestValidationError replaces the built-in response for it. See oxbrook.HTTPError.

Source code in python/oxbrook/_app.py

def exception_handler(self, exc_class: type):
    """Register how an exception becomes a response.

        @app.exception_handler(LookupError)
        async def missing(request, exc):
            return Reply({"error": "not found"}, status=404)

    Applies to the class and its subclasses; the most specific registered
    class wins. Raised by a handler, a dependency, body validation, an
    authorizer or middleware, the exception is mapped before middleware
    sees the result, so the access log records the real status.

    Registering `HTTPError` or `RequestValidationError` replaces the
    built-in response for it. See `oxbrook.HTTPError`.
    """
    if not (isinstance(exc_class, type) and issubclass(exc_class, Exception)):
        raise TypeError(
            f"exception_handler() needs an Exception subclass, got {exc_class!r}"
        )

    def decorator(fn):
        if not inspect.iscoroutinefunction(fn):
            raise TypeError(
                f"exception handler {getattr(fn, '__qualname__', fn)} for "
                f"{exc_class.__name__} must be `async def`"
            )
        existing = self._exception_handlers.get(exc_class)
        if existing is not None:
            raise ValueError(
                f"{exc_class.__name__} already has a handler, "
                f"{getattr(existing, '__qualname__', existing)}"
            )
        self._exception_handlers[exc_class] = fn
        return fn

    return decorator

websocket

websocket(path: str, authorize: Any = None)

Register a WebSocket endpoint.

authorize runs before the handshake and can refuse the upgrade, which the handler cannot: by the time it runs, the 101 has been sent and the client believes it is connected. Return None or True to accept, or a Response/Reply to refuse.

async def members_only(request):
    if not valid(request.header("authorization")):
        return Response(b"nope", status=401, content_type="text/plain")

@app.websocket("/ws", authorize=members_only)
async def feed(request, ws): ...

The handler takes the request and the socket. Oxbrook performs the handshake, so the socket is already open when the handler runs, and the connection closes when it returns.

@app.websocket("/ws")
async def echo(request, ws):
    async for message in ws:
        await ws.send(message)

Source code in python/oxbrook/_app.py

def websocket(self, path: str, authorize: Any = None):
    """Register a WebSocket endpoint.

    `authorize` runs before the handshake and can refuse the upgrade,
    which the handler cannot: by the time it runs, the 101 has been sent
    and the client believes it is connected. Return None or True to
    accept, or a `Response`/`Reply` to refuse.

        async def members_only(request):
            if not valid(request.header("authorization")):
                return Response(b"nope", status=401, content_type="text/plain")

        @app.websocket("/ws", authorize=members_only)
        async def feed(request, ws): ...


    The handler takes the request and the socket. Oxbrook performs the
    handshake, so the socket is already open when the handler runs, and the
    connection closes when it returns.

        @app.websocket("/ws")
        async def echo(request, ws):
            async for message in ws:
                await ws.send(message)
    """

    def decorator(fn):
        route = build_route(fn, "GET", path, websocket=True)
        if authorize is not None:
            route.authorizer = make_gate(authorize)
        self._add(route)
        return fn

    return decorator

backend

backend() -> Any

The shared Redis backend, connected lazily on first use.

Source code in python/oxbrook/_app.py

def backend(self) -> Any:
    """The shared Redis backend, connected lazily on first use."""
    if self._backend is None:
        if not self.redis_url:
            raise RuntimeError(
                "durable topics need a redis_url: App(redis_url='redis://...')"
            )
        from ._redis import RedisBackend

        self._backend = RedisBackend(self.redis_url)
    return self._backend

topic

topic(
    name: str,
    maxsize: int | None = None,
    policy: str | None = None,
    durable: bool = False,
) -> Topic

Get or create a named topic.

Shared across every worker loop in the process, so a message emitted by one handler reaches subscribers running on all of them. durable=True additionally shares it across processes and records it in Redis, which needs App(redis_url=...).

maxsize, policy and durable apply only when the topic is first created.

Source code in python/oxbrook/_app.py

def topic(
    self,
    name: str,
    maxsize: int | None = None,
    policy: str | None = None,
    durable: bool = False,
) -> Topic:
    """Get or create a named topic.

    Shared across every worker loop in the process, so a message emitted by
    one handler reaches subscribers running on all of them. `durable=True`
    additionally shares it across processes and records it in Redis, which
    needs `App(redis_url=...)`.

    `maxsize`, `policy` and `durable` apply only when the topic is first
    created.
    """
    existing = self._topics.get(name)
    if existing is not None:
        return existing
    created = Topic(
        name,
        maxsize=maxsize or 1024,
        policy=policy or DROP_OLDEST,
        backend=self.backend() if durable else None,
    )
    # Racing handlers could both create one; keep whichever landed first so
    # every worker sees the same object.
    return self._topics.setdefault(name, created)

capabilities

capabilities() -> dict[str, Any]

The capabilities this service exposes to agents.

Built from the routes marked tool=True, without starting a server, so it can be inspected or checked into a test.

Source code in python/oxbrook/_app.py

def capabilities(self) -> dict[str, Any]:
    """The capabilities this service exposes to agents.

    Built from the routes marked `tool=True`, without starting a server, so
    it can be inspected or checked into a test.
    """
    from . import _capabilities

    return _capabilities.build(self.routes, self._tool_target)

openapi

openapi() -> dict[str, Any]

The OpenAPI 3.1 document for the routes registered so far.

Built from the same metadata the router uses, so it cannot describe an endpoint the server would not accept. Callable without running the server, which makes it usable for client generation in CI.

Source code in python/oxbrook/_app.py

def openapi(self) -> dict[str, Any]:
    """The OpenAPI 3.1 document for the routes registered so far.

    Built from the same metadata the router uses, so it cannot describe an
    endpoint the server would not accept. Callable without running the
    server, which makes it usable for client generation in CI.
    """
    return _openapi.build(self.routes, self.title, self.version, self.description)

run

run(
    host: str = "127.0.0.1",
    port: int = 8000,
    workers: int | None = None,
    max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    max_body: int = DEFAULT_MAX_BODY,
    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    shutdown_grace: float = DEFAULT_SHUTDOWN_GRACE,
    max_connections: int = DEFAULT_MAX_CONNECTIONS,
    max_message: int = DEFAULT_MAX_MESSAGE,
) -> None

Serve until interrupted.

max_concurrency bounds the requests a single worker loop will accept at once, counting both those queued and those already running. When every worker is at its limit the server answers 503 rather than growing without bound. Lower it for slow handlers, where a deep backlog only adds latency before an inevitable client timeout; raise it to absorb larger bursts of fast requests.

max_body caps a request body; anything larger is answered 413 without being buffered. max_message does the same for a single WebSocket message, where the connection is closed rather than answered.

request_timeout is how long to wait for a handler's first response before answering 504. It does not cut short a stream that has already started, so SSE and WebSocket are unaffected. Zero disables it.

shutdown_grace is how long Ctrl-C waits for in-flight requests to finish before stopping anyway.

max_connections caps sockets held open. At the limit the server stops accepting rather than refusing, so the wait lands in the OS backlog.

Source code in python/oxbrook/_app.py

def run(
    self,
    host: str = "127.0.0.1",
    port: int = 8000,
    workers: int | None = None,
    max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    max_body: int = DEFAULT_MAX_BODY,
    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    shutdown_grace: float = DEFAULT_SHUTDOWN_GRACE,
    max_connections: int = DEFAULT_MAX_CONNECTIONS,
    max_message: int = DEFAULT_MAX_MESSAGE,
) -> None:
    """Serve until interrupted.

    `max_concurrency` bounds the requests a single worker loop will accept
    at once, counting both those queued and those already running. When
    every worker is at its limit the server answers 503 rather than growing
    without bound. Lower it for slow handlers, where a deep backlog only
    adds latency before an inevitable client timeout; raise it to absorb
    larger bursts of fast requests.

    `max_body` caps a request body; anything larger is answered 413 without
    being buffered. `max_message` does the same for a single WebSocket
    message, where the connection is closed rather than answered.

    `request_timeout` is how long to wait for a handler's first response
    before answering 504. It does not cut short a stream that has already
    started, so SSE and WebSocket are unaffected. Zero disables it.

    `shutdown_grace` is how long Ctrl-C waits for in-flight requests to
    finish before stopping anyway.

    `max_connections` caps sockets held open. At the limit the server stops
    accepting rather than refusing, so the wait lands in the OS backlog.
    """
    server = self.build_server(
        host, port, workers, max_concurrency, max_body, request_timeout,
        shutdown_grace, max_connections, max_message, announce=True,
    )
    try:
        server.serve()
    except KeyboardInterrupt:
        pass

build_server

build_server(
    host: str = "127.0.0.1",
    port: int = 8000,
    workers: int | None = None,
    max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    max_body: int = DEFAULT_MAX_BODY,
    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    shutdown_grace: float = DEFAULT_SHUTDOWN_GRACE,
    max_connections: int = DEFAULT_MAX_CONNECTIONS,
    max_message: int = DEFAULT_MAX_MESSAGE,
    announce: bool = False,
)

Prepare a server without starting it.

run uses this; so does the test client, which needs to start the server on one thread and stop it from another. The lifespans run inside serve, not here.

Source code in python/oxbrook/_app.py

def build_server(
    self,
    host: str = "127.0.0.1",
    port: int = 8000,
    workers: int | None = None,
    max_concurrency: int = DEFAULT_MAX_CONCURRENCY,
    max_body: int = DEFAULT_MAX_BODY,
    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
    shutdown_grace: float = DEFAULT_SHUTDOWN_GRACE,
    max_connections: int = DEFAULT_MAX_CONNECTIONS,
    max_message: int = DEFAULT_MAX_MESSAGE,
    announce: bool = False,
):
    """Prepare a server without starting it.

    `run` uses this; so does the test client, which needs to start the
    server on one thread and stop it from another. The lifespans run inside
    `serve`, not here.
    """
    from ._core import Server

    workers = workers or default_workers()
    mode = "GIL" if gil_enabled() else "free-threaded"
    if announce:
        print(
            f"Oxbrook: {workers} worker loop(s), max {max_concurrency} concurrent/worker, "
            f"{mode} Python {sys.version_info.major}.{sys.version_info.minor}"
            f"{', debug' if self.debug else ''}",
            flush=True,
        )
    # Docs first: the document is built from the routes registered so far,
    # so registering /mcp afterwards keeps it out of the OpenAPI paths.
    self._register_docs()
    self._register_mcp()
    specs = [
        (
            r.method,
            r.path,
            # Sockets are left alone: their handshake is already done, so
            # there is nothing for middleware or a mapped status to act on.
            r.target if r.websocket else self._compose(r.target, r.middleware),
            [p.as_spec() for p in r.params],
            r.websocket,
            # The authorizer gets both, so an app-wide auth rule and an
            # `HTTPError(401)` cover sockets even though neither can wrap
            # the socket handler itself.
            None if r.authorizer is None else self._compose(r.authorizer, r.middleware),
            r.stream is not None,
        )
        for r in self.routes
    ]
    lifecycle = Lifecycle(self)
    core = Server(
        host,
        port,
        workers,
        max_concurrency,
        max_body,
        max_message,
        self.debug,
        request_timeout,
        shutdown_grace,
        max_connections,
        not announce,
        specs,
        lifecycle,
        None if self.cors is None else self.cors.as_spec(),
        self._socket_origins(),
    )
    return ServerHandle(core, lifecycle)

Router

Router(prefix: str = '')

A group of routes with a shared prefix and middleware.

Source code in python/oxbrook/_routers.py

def __init__(self, prefix: str = "") -> None:
    self.prefix = check_prefix(prefix)
    self._declared: list[_Declared] = []
    self._middleware: list[Any] = []
    self._children: list[tuple[Router, str]] = []
    self._included = False

prefix instance-attribute

prefix = check_prefix(prefix)

route

route(method: str, path: str, tool: bool = False)

Register a route. See App.route.

Source code in python/oxbrook/_routers.py

def route(self, method: str, path: str, tool: bool = False):
    """Register a route. See `App.route`."""
    method = method.upper()

    def decorator(fn):
        self._declare(_Declared(method, path, fn, tool, False, None))
        return fn

    return decorator

get

get(path: str, tool: bool = False)

Source code in python/oxbrook/_routers.py

def get(self, path: str, tool: bool = False):
    return self.route("GET", path, tool=tool)

post

post(path: str, tool: bool = False)

Source code in python/oxbrook/_routers.py

def post(self, path: str, tool: bool = False):
    return self.route("POST", path, tool=tool)

put

put(path: str, tool: bool = False)

Source code in python/oxbrook/_routers.py

def put(self, path: str, tool: bool = False):
    return self.route("PUT", path, tool=tool)

patch

patch(path: str, tool: bool = False)

Source code in python/oxbrook/_routers.py

def patch(self, path: str, tool: bool = False):
    return self.route("PATCH", path, tool=tool)

delete

delete(path: str, tool: bool = False)

Source code in python/oxbrook/_routers.py

def delete(self, path: str, tool: bool = False):
    return self.route("DELETE", path, tool=tool)

websocket

websocket(path: str, authorize: Any = None)

Register a WebSocket endpoint. See App.websocket.

Source code in python/oxbrook/_routers.py

def websocket(self, path: str, authorize: Any = None):
    """Register a WebSocket endpoint. See `App.websocket`."""

    def decorator(fn):
        self._declare(_Declared("GET", path, fn, False, True, authorize))
        return fn

    return decorator

middleware

middleware(fn)

Middleware for this router's routes only, inside the app's own.

Covers routes on routers included into this one as well. Like app middleware it wraps a socket's authorizer, not the socket handler.

Source code in python/oxbrook/_routers.py

def middleware(self, fn):
    """Middleware for this router's routes only, inside the app's own.

    Covers routes on routers included into this one as well. Like app
    middleware it wraps a socket's authorizer, not the socket handler.
    """
    self._mutable("middleware")
    self._middleware.append(fn)
    return fn

include

include(router: Router, prefix: str = '') -> None

Mount another router under this one.

Source code in python/oxbrook/_routers.py

def include(self, router: "Router", prefix: str = "") -> None:
    """Mount another router under this one."""
    self._mutable(f"router {router!r}")
    if router is self:
        raise ValueError("a router cannot include itself")
    self._children.append((router, check_prefix(prefix)))

HTTPError

HTTPError(
    status: int,
    detail: Any = None,
    headers: dict[str, str] | None = None,
)

Bases: Exception

Raise to answer with an error status.

raise HTTPError(404, "no such user")
raise HTTPError(401, headers={"www-authenticate": "Bearer"})

detail is sent to the client, so it is for messages written for the client. It defaults to the status's standard phrase. This is the one exception whose text is returned: an unhandled exception's never is.

Source code in python/oxbrook/_errors.py

def __init__(
    self,
    status: int,
    detail: Any = None,
    headers: dict[str, str] | None = None,
) -> None:
    if not isinstance(status, int) or not 400 <= status <= 599:
        raise ValueError(
            f"HTTPError needs a 4xx or 5xx status, got {status!r}; "
            f"return a Reply or Response for anything else"
        )
    if detail is None:
        try:
            detail = http.HTTPStatus(status).phrase
        except ValueError:
            detail = "error"
    super().__init__(detail)
    self.status = status
    self.detail = detail
    self.headers = dict(headers or {})

status instance-attribute

status = status

detail instance-attribute

detail = detail

headers instance-attribute

headers = dict(headers or {})

RequestValidationError

RequestValidationError(body: bytes)

Bases: Exception

A request body failed validation. Carries a ready-to-send JSON body.

Source code in python/oxbrook/_schema.py

def __init__(self, body: bytes) -> None:
    super().__init__("request body failed validation")
    self.body = body

body instance-attribute

body = body

errors property

errors: list

The individual failures, for an exception handler that reshapes them.