Tenchi

Tenchi is a small, contract-first Python framework for typed JSON APIs. Contracts define the HTTP boundary, plain async functions implement use cases, and frozen dataclasses carry explicitly wired dependencies.

uv add tenchi

uvx tenchi new my_app
cd my_app
uv sync
uv run tenchi dev

Contracts

A contract declares the method, path, inputs, response body, successful response headers, status, and application errors for an endpoint.

from pydantic import BaseModel, Field
from tenchi.contracts import contract

class CreateTodo(BaseModel):
    title: str = Field(min_length=1)

class Todo(BaseModel):
    id: str
    title: str
    completed: bool

class CreatedTodoHeaders(BaseModel):
    location: str = Field(alias="Location")

create_todo_contract = contract(
    method="POST",
    path="/todos",
    request=CreateTodo,
    response=Todo,
    response_headers=CreatedTodoHeaders,
    status=201,
)

Use cases and ports

Ports are protocols. A frozen context carries their implementations. Use cases are ordinary async functions.

from dataclasses import dataclass
from typing import Protocol

class TodoRepository(Protocol):
    async def create(self, *, title: str) -> Todo: ...
    async def list(self) -> list[Todo]: ...

@dataclass(frozen=True, slots=True)
class AppContext:
    todos: TodoRepository

async def create_todo(request: CreateTodo, context: AppContext) -> Todo:
    return await context.todos.create(title=request.title)

Application wiring

Bind contracts to use cases, compose the routes, and provide concrete dependencies at the application root.

from tenchi.health import health_route
from tenchi.openapi import openapi_route
from tenchi.routes import route, route_group
from tenchi.server import create_app

def create_todo_headers(todo: Todo) -> CreatedTodoHeaders:
    return CreatedTodoHeaders(Location=f"/todos/{todo.id}")

todo_routes = route_group(
    route(
        create_todo_contract,
        create_todo,
        response_headers=create_todo_headers,
    ),
)

routes = route_group(
    todo_routes,
    openapi_route(todo_routes, title="Todos", version="1.0.0"),
    health_route(),
)

app = create_app(
    routes=routes,
    context_factory=lambda: AppContext(todos=repository),
)

The synchronous projector keeps HTTP metadata at the route boundary; Tenchi validates its fixed, scalar fields before the request scope commits. Aliases are the header names on the wire. create_app() also accepts lifespan resources, hooks, middleware, and request-body limits.

Errors

Define stable application errors and declare them on a contract or route group. Undeclared application errors become framework-owned 500s.

from tenchi.errors import AppError, ErrorDef

todo_not_found = ErrorDef(
    code="TODO_NOT_FOUND",
    status=404,
    message="Todo not found",
)

get_todo_contract = contract(
    method="GET",
    path="/todos/{todo_id}",
    params=GetTodoParams,
    response=Todo,
    errors=(todo_not_found,),
)

async def get_todo(params: GetTodoParams, context: AppContext) -> Todo:
    todo = await context.todos.get(params.todo_id)
    if todo is None:
        raise AppError(todo_not_found, details={"todo_id": params.todo_id})
    return todo

Authentication belongs in boundary hooks. Authorization stays in use cases and pure policy functions.

Client and OpenAPI

The async client serializes inputs and validates responses using the same contract.

from tenchi.client import Client

async with Client(base_url="https://api.example.com") as client:
    created = await client.call_with_response(
        create_todo_contract,
        request=CreateTodo(title="Buy milk"),
    )
    todo = created.body
    location = created.headers.location
    status = created.http_response.status_code

call() still returns only the validated body. Both client methods validate declared success headers. openapi_route() serves an OpenAPI 3.1 document from the same declarations, including the headers.

The taskboard example builds optimistic concurrency on this surface: task responses emit ETag, PATCH requests require If-Match, and stale writes return a declared 412. Task creation also requires a typed Idempotency-Key so concurrent retries create once, replay the original response, and reject a reused key with different input as a declared 409.

Outcomes, deadlines, and observation

Routes that can succeed in more than one way declare named outcomes. A synchronous presenter selects the status-specific body and headers while the use case keeps returning domain data.

from tenchi.responses import PresentedResponse, present, success

created = success(name="created", status=201, response=Todo)
existing = success(name="existing", status=200, response=Todo)

put_todo_contract = contract(
    method="PUT",
    path="/todos",
    request=CreateTodo,
    response=Todo,
    successes=(created, existing),
    timeout=5.0,
)

def present_put(result: PutTodoResult) -> PresentedResponse:
    outcome = created if result.created else existing
    return present(outcome, body=result.todo)

route(put_todo_contract, put_todo, present=present_put)

A success declared with passthrough=True may instead carry a Starlette streaming, file, or redirect response. Tenchi preserves the response while checking its contract-owned status, media-type parameters, and declared headers. Outcomes without a body accept only a concrete, provably empty response; streaming outcomes declare their body type. The typed client validates the selected outcome and exposes it as ClientResponse.success.

timeout= cooperatively cancels overdue work, waits for request-scope cleanup, and returns a framework 504 even if application code catches the cancellation. create_app(observers=...) delivers immutable outcomes with read-only request headers after matched requests finalize for metrics or tracing; observer failures are logged and isolated from the response.

Workers and scripts

execute() validates job data against a use case’s request annotation and applies the same context scoping as the server.

from tenchi.execution import execute

await execute(
    notify_member_added,
    request_json=job.payload,
    context=context,
)

Pagination

Subclass PageQuery for filters, return Page[Item], and build the result with page().

from tenchi.pagination import Page, PageQuery, page

class ListTodosQuery(PageQuery):
    completed: bool | None = None

async def list_todos(
    query: ListTodosQuery,
    context: AppContext,
) -> Page[Todo]:
    items = await context.todos.list()
    start = query.offset
    window = items[start : start + query.limit]
    return page(window, total=len(items), query=query)

Testing

open_client provides the typed client in-process; open_http exposes raw httpx responses. Both run the app lifespan.

from tenchi.testing import open_client, open_http

async with open_client(app) as client:
    todo = await client.call(
        create_todo_contract,
        request=CreateTodo(title="Buy milk"),
    )

async with open_http(app) as http:
    response = await http.get("/health")
    assert response.status_code == 200

CLI

tenchi new my_app
tenchi make feature notes
tenchi make use-case notes create_note
tenchi routes
tenchi openapi
tenchi doctor
tenchi dev