BlackBull

BlackBull is a Python web framework — native typed Connection end to end, ASGI 3.0 kept as an interop boundary — for developers who want one pip install, zero C compilers, and the ability to programmatically test their HTTP clients and servers against deliberate protocol misbehaviour.

Live demo → — BlackBull running in production on a free-tier host, no external server in front of it.

Hello, world

from blackbull import BlackBull

app = BlackBull()

@app.route(path='/')
async def hello():
    return "Hello, world!"

if __name__ == '__main__':
    app.run(port=8000)

$ python app.py &
$ curl -i http://localhost:8000/
HTTP/1.1 200 OK
content-type: text/plain; charset=utf-8
content-length: 13

Hello, world!

Why BlackBull

• Zero ceremony. app.run() is the entire deploy story — or blackbull serve ./public for a static site with ETag + HTTP/2 and no code at all. No separate ASGI runner, no YAML config, no gunicorn class path.
• Readable stack. Every byte on the wire passes through Python you can step through with pdb. No C extensions to debug.
• New protocols, early. HTTP QUERY (RFC 10008) — a safe, idempotent, cacheable method with a request body — ships with routing, Accept-Query negotiation, and Content-Type enforcement, years before it reaches the standard library.
• Declare, don't plumb. Handlers name what they need — path params, query params, the body, a Connection view, a WebSocket on a websocket route, or a Depends(get_db) resource with teardown when the handler is done — and the router resolves it all when the route is registered. The same signature means the same thing on an HTTP route and a WebSocket one. A handler that uses none of it compiles to the same bare wrapper: zero per-request cost for features you didn't ask for.
• Break things on purpose. The same protocol code that serves real traffic can drive a programmable misbehaving client or server. Test your own HTTP/2 client against half-closed streams, exhausted windows, and illegal SETTINGS — in CI.
• Test the path your requests actually take. blackbull.testing.native calls your app the way the server does — a typed Connection, no ASGI scope round-trip — and NativeTestServer runs the whole stack on a loopback port when the answer depends on the wire (keep-alive, HEAD, chunked framing). Neither needs a subprocess. See docs/guide/testing.md.
• Multi-protocol, one process. A pure-Python MQTT 5 broker and gRPC ride beside HTTP/2 and WebSocket on the same runtime; extensions add new protocols without touching the core — still no C extension. That combination is an edge inference API server in one python app.py: SSE token streaming with HTTP/2 multiplexing for clients, MQTT ingest and $share/… work queues for devices — on any box CPython runs on, ARM included.
• Actor-model internals. Connection-level isolation without a single shared lock: a ConnectionActor spawns a per-connection protocol actor whose inbox loop owns its own state. The same message-passing concurrency runs your HTTP routes, the MQTT broker, and the gRPC handlers — one model across every protocol.
• RFC-grade correctness. Passes the same external conformance suites used to validate nginx and Envoy (h2spec, Autobahn). First Python framework with native HTTP QUERY (RFC 10008) support — the new safe, idempotent, cacheable method that carries a request body, with Accept-Query content negotiation.
• Typed throughout. Your editor and mypy / pyright see every parameter; PEP 561 typed distribution. The receive/send message channel is a discriminated union — all 19 ASGI event shapes are TypedDicts keyed by a Literal tag, so event['type'] narrows and a malformed send is a type error, not a traceback.

Install

pip install blackbull
pip install 'blackbull[compression]'      # add brotli + zstandard codecs
pip install 'blackbull[speed]'            # add uvloop event loop
pip install 'blackbull[reload]'           # add watchfiles for --reload
pip install 'blackbull[fault-injection]'  # add cryptography + httpx for the toolkit
pip install 'blackbull[mqtt]'             # MQTT 5 broker extension
pip install 'blackbull[grpc]'             # gRPC over HTTP/2 (all four RPC shapes)
pip install 'blackbull[protobuf]'         # + protobuf servicers, reflection, health, rich errors

Simplified handlers

Route handlers may return a str, bytes, dict, or Response; path parameters are coerced to the annotation type:

@app.route(path='/tasks/{task_id:int}')
async def get_task(task_id: int):
    return {"id": task_id, "title": "..."}

Need more than the URL? Declare a Connection parameter — headers, cookies, client, and body()/json()/text(), injected only for handlers that ask for it:

@app.route(path='/notes/{note_id:int}', methods=[HTTPMethod.POST])
async def create_note(note_id: int, conn: Connection):
    return {"id": note_id, "note": await conn.json()}

Query params and per-request resources are declared the same way — everything is resolved when the route is registered, so handlers that don't use a feature pay nothing for it at request time:

from blackbull import Depends

async def get_db():
    conn = await pool.acquire()
    try:
        yield conn                 # injected value
    finally:
        await pool.release(conn)  # runs after the response is sent

@app.route(path='/search')
async def search(q: str, page: int = 1, db=Depends(get_db)):
    return await db.find(q, page=page)   # /search?q=bull&page=2

Drop down to the full (conn, receive, send) form whenever you need it — routes accept either shape. conn is the native typed Connection; receive/send carry ASGI 3.0 events, declared as TypedDicts so a type checker narrows them on event['type'].

TLS + HTTP/2

app.run(port=8443, certfile='cert.pem', keyfile='key.pem')

ALPN negotiates h2 automatically; HTTP/1.1 clients fall back via the same socket.

HTTP QUERY (RFC 10008)

A safe, idempotent, cacheable method that carries a request body — for the searches that don't fit in a URL and shouldn't be a POST:

from blackbull import QUERY

@app.route(path='/search', methods=[QUERY])
async def search(body: bytes):
    return run_query(body)

BlackBull ships QUERY routing, Accept-Query content negotiation, and Content-Type enforcement. http.HTTPMethod has no QUERY member — RFC 10008 postdates the 3.15 feature freeze, so the earliest stdlib arrival is 3.16 — and because HTTPMethod is a StrEnum, the exported string stays equal- and hash-compatible with a future HTTPMethod.QUERY. Routes registered today need no migration.

See the routing guide.

WebSocket

from blackbull import WebSocket
from blackbull.utils import Scheme

@app.route(path='/ws', scheme=Scheme.websocket)
async def ws_echo(ws: WebSocket):
    await ws.accept()
    async for message in ws:
        await ws.send(message)

The loop ends when the client disconnects. send_text / send_bytes / send_json and receive_text / receive_bytes / receive_json are there when you want to be explicit, and await ws.close(code, reason) before accept() rejects the handshake outright.

A WebSocket handler declares what it needs the same way an HTTP one does — path params, query params, and Depends all resolve from the signature:

@app.route(path='/rooms/{room}', scheme=Scheme.websocket)
async def chat(ws: WebSocket, room: str, since: int = 0, db=Depends(get_db)):
    await ws.accept()
    ...

A Depends on a socket is resolved once per connection and released when the handler exits, so give scarce resources application scope and borrow them per use — the guide explains why. The raw (conn, receive, send) event form is still supported — see WebSockets.

Beyond HTTP — the Non-ASGI bridge

One process, more than one protocol. Extensions attach non-HTTP protocols through a single seam — app.add_extension(...) — while the HTTP core stays protocol-agnostic. The flagship is a pure-Python MQTT 5 broker: CONNECT / SUBSCRIBE / PUBLISH at QoS 0–2, retained messages, Last-Will, and shared subscriptions ($share/… work queues) on the standard :1883 port — or over TLS with MQTTExtension(port=8883, tls=True) — beside your HTTP routes, no Mosquitto sidecar, no C extension.

from blackbull import BlackBull
from blackbull.mqtt import MQTTExtension, Message

app = BlackBull()
mqtt = app.add_extension(MQTTExtension(port=1883))

@mqtt.on_message(topic='sensors/{room}/temperature')
async def on_temp(msg: Message, room: str):
    print(room, msg.payload.decode())   # {room} captured like an HTTP path param

app.run(port=8000)   # HTTP on 8000, MQTT on 1883

Need a protocol BlackBull doesn't ship? @app.raw_handler hands you the raw reader/writer for a port. See docs/guide/mqtt.md and docs/guide/raw-protocols.md.

Deeper dives: Architecture · Is BlackBull right for you? · Known limitations

Event API

Two decorators cover both lifecycle hooks and per-request behaviour — observation (@app.on, fire-and-forget) and interception (@app.intercept, synchronous, may short-circuit):

@app.on_startup
async def warm_caches():
    ...

@app.intercept('request_received')
async def auth(scope, receive, send, call_next):
    # raise to abort, or skip call_next to short-circuit
    await call_next(scope, receive, send)

@app.on('request_completed')
async def emit_metrics(event):
    metrics.increment('requests', status=event['status'])

Events: app_startup, app_shutdown, request_received, before_handler, request_completed, websocket_message. @app.on isolates exceptions per observer; @app.intercept is part of the request path and can deny / rewrite / pass through. See docs/guide/events.md for the full event catalogue and detail payloads.

Built-in middleware

Compose via app.use(...) or per-route middlewares=[...]:

OpenAPI / Swagger UI

app.enable_openapi()   # publishes /openapi.json and /docs

Auto-generates an OpenAPI 3.1 spec from route signatures, path-param converters, docstrings, and @dataclass annotations on body parameters. Dataclass-typed bodies are also deserialized at runtime — async def h(body: CreateTask): ... receives a constructed instance, no manual json.loads.

For the MQTT broker there is a messaging counterpart: AsyncAPIExtension publishes an AsyncAPI 3.0 document for your @mqtt.on_message taps at /asyncapi.json (with an HTML viewer at /asyncapi). See docs/guide/mqtt.md.

Fault injection

BlackBull's single most distinctive feature: a programmable deliberate-misbehaviour toolkit you can point at your own HTTP/2 client (or proxy, or middleware) directly from a pytest suite.

import pytest
from blackbull.fault_injection import H2FaultServer, make_self_signed_h2_context
from blackbull.fault_injection.catalogue import half_closed_stream_no_data

@pytest.mark.asyncio
async def test_my_client_handles_half_closed_streams():
    ssl_ctx = make_self_signed_h2_context()
    async with H2FaultServer(
        scenario=half_closed_stream_no_data(), ssl_context=ssl_ctx,
    ) as srv:
        # Your client must time out or RST_STREAM rather than block
        # forever when the server sends HEADERS without END_STREAM.
        with pytest.raises(TimeoutError):
            await my_h2_client.get(srv.url, timeout=1.0)

The catalogue ships four spec-grade categories (half-closed streams, exhausted flow-control windows, illegal SETTINGS, weird frame sequences); the symmetric HTTP/1.1 client side drives a real server through trickled headers, partial requests, and abrupt RST. See docs/guide/fault_injection.md for the full tutorial.

Early Alpha

⚠ Early Alpha — The API may change between MINOR versions. See Conformance for protocol-level test coverage and Known Limitations for the explicit list of behaviours to expect before adopting.

Examples

Documentation

• Guide: docs/guide/index.md
• Architecture & design rationale: docs/about/architecture.md
• Implementation internals: docs/about/internals.md
• Changelog: CHANGELOG.md

Versioning

BlackBull uses ZeroVer prior to a 1.0 commitment. MINOR advances at each sprint close; PATCH is for bug fixes and harness work between sprints. See CHANGELOG.md for the full release history.

License

Apache License 2.0 — © TOKUJI.

Next steps

→ Read the Guide — routing, middleware, WebSockets, HTTP/2, and more.

→ Browse the examples — copy-pasteable starting points for REST APIs, chat servers, and SSE streams.

→ Star on GitHub — every star helps the project grow.