mxhttp

Declarative HTTP client on top of msgspec and httpx. Write an API as a class of annotated stub methods and mxhttp will handle the rest (request building, sending, and response decoding).

Install

pip install mxhttp

Usage

from typing import Annotated
import msgspec
from mxhttp import Body, Query, SyncConsumer, get, post


class Item(msgspec.Struct):
    id: int
    name: str
    price: float


class NewItem(msgspec.Struct):
    name: str
    price: float


class Shop(SyncConsumer):
    @get("/items/{item_id}")
    def get_item(self, item_id: int) -> Item: ...  # type: ignore[empty-body]

    @get("/search")
    def search(self, q: Annotated[str, Query], limit: Annotated[int, Query] = 20) -> list[Item]: ...  # type: ignore[empty-body]

    @post("/items")
    def create_item(self, item: Annotated[NewItem, Body]) -> Item: ...  # type: ignore[empty-body]


shop = Shop("https://api.example.com")
item = shop.get_item(item_id=7)
new = shop.create_item(item=NewItem(name="Gadget", price=4.5))

The method body is never run as it is replaced by the decorator. Parameters are bound based on their Annotated[...] marker:

• Use Path["name"], Query["name"], Field["name"], Header["name"], or Cookie["name"] to bind under a different name than the parameter (e.g. reserved from, or a header like X-Request-Id, unsupported string format arguments like ?).
• None-valued Query, Field, Header, and Cookie parameters are omitted from the request.
• Path parameters cannot be optional as a placeholder cannot be ommited from the URL.
• Mismatched marker/type combinations raise a TypeError as soon as the class body runs, not at call time.

Decoding the response

The return type defines the reponse decoding:

• httpx.Response for the raw response.
• str or bytes for the corresponding .text or .content with no JSON round-trip.
• pydantic.BaseModel subclasses via their own .model_validate_json.
• Anything else msgspec.json.decode can decode: msgspec.Struct, dataclasses, TypedDict, NamedTuple, and list, dict, or other containers of those.
• Response[Item] for a small struct with the decoded Item as .data and the raw httpx.Response in .response.
• Plain attrs classes are decoded by msgspec, for type hinting attrs is needed as dependency.

For an async client, subclass AsyncConsumer and declare the methods async def, everything else stays the same.

Response handling

By default, every response is checked by response.raise_for_status() before decoding, so errors during the request raise httpx.HTTPStatusError automatically. This behavior can be overriden by @response_handler decorator for the class.

import httpx
from mxhttp import response_handler


def ignore_errors(response: httpx.Response) -> httpx.Response:
    return response


@response_handler(ignore_errors)
class Shop(SyncConsumer): ...

The hook runs on every response before decoding.

Streaming responses

Annotate the return type as Iterator[bytes] (sync) or AsyncIterator[bytes] (async) to stream the response body in chunks.

from collections.abc import AsyncIterator, Iterator


class Files(SyncConsumer):
    @get("/files/{file_id}")
    def download(self, file_id: int) -> Iterator[bytes]: ...  # type: ignore[empty-body]


for chunk in shop_files.download(file_id=7):
    ...

class AsyncFiles(AsyncConsumer)
    @get("/files/{file_id}")
    def download(self, file_id: int) -> AsyncIterator[bytes]: ...  # type: ignore[empty-body]

async for chunk in await shop_async_files.download(file_id=7):
    ...

httpx already decompresses chunks before responding according to Content-Encoding (gzip/deflate/br/zstd).

Streaming responses run @streaming_response_handler instead of @response_handler (defaults to raise_for_status as well). The handler can only inspect status line and headers.

from mxhttp import streaming_response_handler


def check_status(response: httpx.Response) -> httpx.Response:
    response.raise_for_status()
    return response


@streaming_response_handler(check_status)
class Files(SyncConsumer): ...

Server-Sent Events

Annotate the return type as Iterator[Event] (sync) or AsyncIterator[Event] (async) to parse the response as a Server-Sent Events stream instead of raw bytes:

from collections.abc import Iterator
from mxhttp import Event


class Chat(SyncConsumer):
    @get("/stream")
    def events(self) -> Iterator[Event]: ...  # type: ignore[empty-body]


for event in chat.events():
    print(event.event, event.data)  # event.event defaults to "message"

Event has four attributes, data, event, id, and retry:

• data is the raw payload, decode it manually if the server sends JSON.
• Multi-line data fields are joined with \n.
• id and retry persist across events once set and reset on reconnect only.
• An event without a trailing blank line at the end of the stream is discarded.

SSE streams use @streaming_response_handler matching byte streaming above.

Further configuration

The underlying httpx.Client or httpx.AsyncClient is stored at .session to set default headers, auth, or timeouts.

Typing

The package and all its generators are typed.

Tests

pytest

Acknowledgements

mxhttp is inspired by Uplink but combining it with Python typing features.