# Tenchi
Source: https://tenchi.io

Tenchi

  Tenchi is ready for evaluation and real application feedback, but its public
  API can still change between minor releases. Read [stability and
  releases](/stability) before adopting it for a long-lived service.

A contract-first, Python-native framework for typed JSON APIs and AI-facing
tools that stay coherent as they grow.

Tenchi gives the HTTP boundary, application tools, use cases, a typed Python
client, OpenAPI, compatibility checks, and structural tooling one shared
model. Contracts describe what crosses the wire. Plain async functions own
behavior. Protocols and frozen dataclasses keep dependencies explicit.

## Start an application

```shell
uvx tenchi new my_app
cd my_app
uv sync
uv run tenchi check
uv run tenchi dev
```

The generated application includes a working todos feature, SQLite runtime
persistence, a memory test adapter, Swagger UI, strict Pyright configuration,
OpenAPI and application-tool snapshots, an agent guide, and CI. The
[quickstart](/getting-started) follows its first development loop.

## What makes it different

- **The contract is executable.** The server validates requests and responses,
  the client validates both sides of the call, and OpenAPI is derived from the
  same declaration.
- **Application code stays ordinary.** Use cases are plain async functions;
  ports are `typing.Protocol`; contexts are frozen dataclasses.
- **API changes are reviewable.** Canonical snapshots and conservative
  compatibility analysis distinguish additive, breaking, metadata-only, and
  unknown changes.
- **Architecture is explicit.** `tenchi doctor` checks dependency direction
  without introducing a dependency-injection container or framework base
  classes.
- **Tooling is agent-readable.** Versioned maps, mutation previews, structured
  diagnostics, and one complete check command give coding agents the same
  evidence used by people and CI.
- **HTTP is one entrypoint.** Workers and scripts can execute the same use cases
  with the same request validation and context-scoping guarantees.
- **AI tools reuse application behavior.** Typed tool declarations add
  deterministic schemas, declared errors, and safety metadata around ordinary
  use cases without choosing a model provider or agent loop.

## One declaration, several guarantees

```python
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

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

That value can be bound to a use case, called through the typed client, served
as OpenAPI 3.1, and compared against a previous release.

## Read next

- [Quickstart](/getting-started) creates and changes a complete application.
- [Add Tenchi to an existing project](/existing-project) builds the first
  operation without using the application generator.
- [Mental model](/concepts) defines the small vocabulary used by the framework.
- [Contracts](/contracts) covers the complete HTTP declaration.
- [Production handbook](/production) covers configuration, transactions,
  retries, background work, observability, and deployment.
- [Signed webhooks](/webhooks) verifies provider deliveries before parsing and
  makes redeliveries safe.
- [Rate limiting](/rate-limits) applies authenticated operation quotas without
  coupling use cases to Redis, a database, or an API gateway.
- [Coding agents](/agents) defines the inspect, preview, edit, and validate
  protocol for automated changes.
- [Application tools](/tools) exposes existing use cases to AI and machine
  callers through a validated, transport-neutral boundary.
- [Application MCP](/tool-mcp) publishes those tools with authenticated
  discovery, explicit approval for destructive calls, and structured results.
- [Comparisons](/comparisons) explains where Tenchi fits and where it does not.

---

# Quickstart
Source: https://tenchi.io/getting-started

Create a Tenchi application, run its checks, and make one contract change that
you can observe at every boundary.

## Requirements

Tenchi requires Python 3.12 or newer. The examples use
[uv](https://docs.astral.sh/uv/) to create environments and run commands.

## Create the application

```shell
uvx tenchi new my_app
cd my_app
uv sync
```

The scaffold is intentionally a small complete application rather than a
single route. Its important files are:

| Path | Responsibility |
| --- | --- |
| `app/features/todos/contracts.py` | HTTP method, path, inputs, response, and metadata |
| `app/features/todos/schemas.py` | Pydantic boundary and application data models |
| `app/features/todos/use_cases/` | Plain application functions |
| `app/features/todos/ports.py` | App-owned infrastructure protocols |
| `app/infra/` | SQLite runtime adapter, memory test adapter, and port wiring |
| `app/server/context.py` | Frozen context passed to use cases |
| `app/server/runtime.py` | Lifespan and context wiring shared by entrypoints |
| `app/server/preflight.py` | Read-only checks of the target deployment environment |
| `app/server/jobs.py` | Background-job handler composition |
| `app/server/tasks.py` | Operational task composition |
| `app/server/tools.py` | Registered machine-facing tool composition |
| `app/server/asgi.py` | Composition root and ASGI application |
| `openapi.json` and `tools.json` | Version-controlled HTTP and tool contracts |
| `AGENTS.md` | Placement rules and the complete agent validation loop |
| `.mcp.json` | Project-local registration for MCP-aware coding agents |
| `.github/workflows/ci.yml` | Checks and historical boundary compatibility gates |

The starter preflight group is empty because a local source check must not
contact an environment implicitly. Add production dependency observations by
following [deployment preflight](/preflight).

## Run the checks

```shell
uv run tenchi check
```

The command runs Ruff formatting and linting, Pyright, pytest, doctor, and the
OpenAPI and application-tool snapshot checks. It runs every step even after a
failure so one pass reports the complete repair list. Generated code passes
untouched. Use-case tests run without an HTTP server; integration tests
exercise the composed ASGI application.

## Start the server

```shell
uv run tenchi dev
```

The runtime database defaults to `my_app.db`. Set `MY_APP_DATABASE` to choose a
different path.

In another terminal:

```shell
curl -i \
  -H 'content-type: application/json' \
  -d '{"title":"Buy milk"}' \
  http://127.0.0.1:8000/todos

curl http://127.0.0.1:8000/todos
```

The create response is validated as `Todo`, returns status `201`, and includes
the declared `Location` header. Restart the server and list again: the todo is
still present because the startup lifespan ensures the SQLite schema while
each request owns its own commit-or-rollback transaction.

Open [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs) to inspect and call
the same contracts through Swagger UI. The underlying document is served at
`/openapi.json`, and `/health` exposes the starter's liveness route.

## Make the first contract change

Open `app/features/todos/schemas.py` and require a longer title:

```diff
 class CreateTodo(BaseModel):
-    title: str = Field(min_length=1)
+    title: str = Field(min_length=3)
```

The server now rejects shorter input before the use case runs. Generate the
current OpenAPI and compare it to the committed baseline:

```shell
uv run tenchi openapi \
  --routes app.server.routes:api_routes \
  --title my_app \
  --diff openapi.json
```

Tightening a request constraint is breaking for existing callers, so the
command fails and explains the affected operation. Restore the old constraint,
then add an optional field instead:

```python
class CreateTodo(BaseModel):
    title: str = Field(min_length=1)
    notes: str | None = None
```

The compatibility report now classifies the change as additive. After
reviewing it, update the snapshot:

```shell
uv run tenchi openapi \
  --routes app.server.routes:api_routes \
  --title my_app \
  --write openapi.json
uv run tenchi check
```

  Run `--diff` before replacing the snapshot. The generated CI runs
  `--diff-ref` against the pull request's base commit, so committing an updated
  snapshot cannot hide an incompatible contract change.

## Add another feature

```shell
uv run tenchi make feature notes
uv run tenchi make use-case notes create_note
uv run tenchi map --feature notes
```

Generators create files and print the wiring steps. They do not edit existing
composition modules, keeping dependency wiring visible and app-owned.
Add `--dry-run` to preview every file or `--json` for a versioned result.
The map shows the new declarations before they are wired. Runtime-bound
contracts, use cases, and adapters become registered after you follow the
printed composition steps. Use
`tenchi map --feature notes --json` when an agent or another tool will consume
the graph. See [coding agents](/agents) for the complete inspect, preview, edit,
and validate loop, or [connect the coding-agent MCP server](/mcp) to expose
those operations as agent tools.

Next, read the [mental model](/concepts), then follow [app
architecture](/architecture) when deciding where new code belongs.

---

# Add to an existing project
Source: https://tenchi.io/existing-project

Use this path when you already have a Python project and want to add a Tenchi
API without starting from the generated application. The result is one working
JSON operation, a direct use-case test, OpenAPI and health routes, and the same
complete validation command used by generated projects.

## Install Tenchi and development tools

Tenchi requires Python 3.12 or newer. From the project root, add the runtime and
local development dependencies:

```shell
uv add tenchi
uv add --dev uvicorn ruff pyright pytest pytest-asyncio
```

Create this initial structure. Empty `__init__.py` files make each package
explicit:

```text
app/
  __init__.py
  features/
    __init__.py
    greetings/
      __init__.py
      contracts.py
      routes.py
      schemas.py
      use_cases/
        __init__.py
        greet.py
      tests/
        __init__.py
        test_greet.py
  server/
    __init__.py
    asgi.py
    context.py
    jobs.py
    preflight.py
    routes.py
    runtime.py
    tasks.py
    tools.py
```

## Declare the boundary

Define the validated query and response in
`app/features/greetings/schemas.py`:

```python
from pydantic import BaseModel, Field

class GreetQuery(BaseModel):
    name: str = Field(min_length=1)

class Greeting(BaseModel):
    message: str
```

Describe the HTTP operation in `app/features/greetings/contracts.py`:

```python
from tenchi.contracts import contract

from .schemas import Greeting, GreetQuery

greet_contract = contract(
    method="GET",
    path="/greet",
    query=GreetQuery,
    response=Greeting,
    name="greet",
    summary="Greet someone",
    public=True,
)
```

`public=True` gives a future authentication hook an explicit exemption signal.
It does not change access by itself.

## Implement and bind the use case

Create the application context in `app/server/context.py`. This first operation
has no external dependencies, so the context is empty:

```python
from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class AppContext:
    pass
```

Implement the behavior in `app/features/greetings/use_cases/greet.py`:

```python
from app.server.context import AppContext

from ..schemas import Greeting, GreetQuery

async def greet(query: GreetQuery, context: AppContext) -> Greeting:
    return Greeting(message=f"Hello, {query.name}!")
```

Bind the contract to the use case in `app/features/greetings/routes.py`:

```python
from tenchi.routes import route, route_group

from .contracts import greet_contract
from .use_cases.greet import greet

routes = route_group(route(greet_contract, greet))
```

`route()` checks the query and return annotations immediately. Importing this
module fails if the use-case signature no longer matches the contract.

## Compose the ASGI application

Compose the application and documentation routes in `app/server/routes.py`:

```python
from tenchi.health import health_route
from tenchi.openapi import openapi_route, swagger_ui_route
from tenchi.routes import route_group

from app.features.greetings.routes import routes as greeting_routes

OPENAPI_TITLE = "Existing app"
OPENAPI_VERSION = "0.1.0"
OPENAPI_DESCRIPTION = "Greeting API"

api_routes = route_group(greeting_routes)

routes = route_group(
    api_routes,
    openapi_route(
        api_routes,
        title=OPENAPI_TITLE,
        version=OPENAPI_VERSION,
        description=OPENAPI_DESCRIPTION,
    ),
    swagger_ui_route(title=f"{OPENAPI_TITLE} documentation"),
    health_route(),
)
```

Put entrypoint-neutral context wiring in `app/server/runtime.py`:

```python
from app.server.context import AppContext

def create_context() -> AppContext:
    return AppContext()
```

Expose the Starlette application from `app/server/asgi.py`:

```python
from tenchi.server import create_app

from app.server.routes import routes
from app.server.runtime import create_context

app = create_app(routes=routes, context_factory=create_context)
```

When the application needs a database or SDK client, replace the direct context
factory with the lifespan and request-scope pattern from [routes and
server](/server).

Create an empty operational runner in `app/server/tasks.py`:

```python
from tenchi.tasks import create_task_runner, task_group

from app.server.runtime import create_context

tasks = task_group()
runner = create_task_runner(tasks=tasks, context_factory=create_context)
```

This keeps `tenchi map` and `tenchi task list` useful before the application
has a maintenance command. Add tasks later by following [Operational
tasks](/tasks).

Create an empty background-job composition in `app/server/jobs.py`:

```python
from tenchi.jobs import job_group

jobs = job_group()
```

This gives `tenchi map` its conventional job target before the application has
durable work. Add producer and consumer bindings later by following
[Background jobs](/jobs).

Create an empty application-tool composition in `app/server/tools.py`:

```python
from tenchi.tools import tool_group

tools = tool_group()
```

This gives `tenchi map`, `tenchi tools`, and `tenchi check` their conventional
target before the application exposes a use case to machine callers. Add
declarations and authenticated runner wiring later by following [Application
tools](/tools).

Create an empty deployment gate in `app/server/preflight.py`:

```python
from tenchi.preflight import preflight_group

checks = preflight_group()
```

The empty group lets `tenchi preflight` pass before the application has an
external dependency. Add read-only database, secret-manager, service, or
worker observations by following [deployment preflight](/preflight).

## Add a direct behavior test

Create `app/features/greetings/tests/test_greet.py`:

```python
from app.features.greetings.schemas import GreetQuery
from app.features.greetings.use_cases.greet import greet
from app.server.context import AppContext

async def test_greet() -> None:
    result = await greet(GreetQuery(name="Tenchi"), AppContext())

    assert result.message == "Hello, Tenchi!"
```

Merge these settings into `pyproject.toml` so the aggregate check knows where
to find the application and tests:

```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["app"]
pythonpath = ["."]

[tool.ruff]
line-length = 88

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM", "RUF"]

[tool.pyright]
include = ["app"]
typeCheckingMode = "strict"
pythonVersion = "3.12"
```

Keep any existing test and source paths when merging these tables into the
project's configuration.

## Create the baseline and verify the app

Write the first canonical OpenAPI and application-tool snapshots, then run the
complete project gate:

```shell
uv run tenchi openapi \
  --routes app.server.routes:api_routes \
  --title "Existing app" \
  --version 0.1.0 \
  --description "Greeting API" \
  --write openapi.json
uv run tenchi tools --write tools.json
uv run tenchi check
uv run tenchi preflight
uv run tenchi task list
```

Start the development server after the checks pass:

```shell
uv run tenchi dev
```

Open [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs), or call the
operation directly:

```shell
curl "http://127.0.0.1:8000/greet?name=Tenchi"
```

The response is `{"message":"Hello, Tenchi!"}`. Continue with [app
architecture](/architecture) before adding ports, adapters, policies, and
authenticated operations.

---

# Mental model
Source: https://tenchi.io/concepts

Tenchi has one architecture: typed contracts at the boundary, plain use cases
at the center, and explicit dependency wiring around them.

## Contract

A frozen declaration of one HTTP operation: method, path, validated inputs,
successful responses, expected application errors, media types, metadata, and
runtime limits. A contract contains no handler logic.

## Route

A binding between a contract and a use case. `route()` inspects the use-case
signature immediately, so mismatched parameter names or annotations fail while
the application is composed rather than on the first request.

## Use case

A plain async function that implements one application action. Its parameters
are named boundary values such as `request`, `params`, `query`, `headers`, and
`context`. It does not receive a Starlette request.

## Operational task

A stable name bound to a use case for a backfill, repair, replay, or maintenance
command. A task validates its input and result and runs with application
lifecycle resources. It is an invocation boundary, not a scheduler or queue.

## Background job

A durable, stable name with validated request and result types. Producers
serialize a `JobMessage`; the composition root binds the declaration to a plain
async consumer; a dispatcher validates delivery. Queue persistence, retries,
acknowledgement, and scheduling stay in infrastructure.

## Application tool

A stable machine-facing name with validated input, output, declared errors,
and safety annotations, explicitly bound to a plain async use case. A tool
runner supplies application lifecycle and context wiring. Model providers,
agent loops, authentication, approval interfaces, and transport protocols stay
outside the declaration.

## Port

An interface the application needs, declared with `typing.Protocol`. The
feature owns the port; infrastructure supplies an adapter. This keeps the
application independent of a database driver, HTTP SDK, cache, or queue.

## Context

A frozen dataclass carrying the dependencies and verified identity available
to a use case. The application constructs it explicitly. Tenchi does not use a
service locator or dependency-injection container.

## Hook

An HTTP-boundary function that sees `RequestInfo` and the current context. A
hook can authenticate a request and return an enriched context. Hooks are for
boundary concerns, not business authorization.

## Policy

A pure application function that answers an authorization question about a
known subject. The use case loads the subject through a port and asks the
policy. Because policies perform no I/O, they are easy to test and reusable
from HTTP, workers, scripts, and tests.

## Adapter

A concrete implementation of a port. SQL repositories, external API clients,
and memory test doubles are adapters. They live under `app/infra/` and are
wired at the server composition root.

## Boundary flow

```text
HTTP request
  -> route match
  -> request id and context factory / request scope
  -> boundary hooks
  -> size, media-type, and Pydantic input validation
  -> plain use case
  -> response presenter and header projector
  -> Pydantic response validation
  -> HTTP response
```

The typed client runs the complementary checks: it serializes inputs from the
same contract and validates the status, media type, body, headers, and declared
application errors returned by a server.

## The key tradeoff

Tenchi asks for more structure than a one-file microframework example. In
return, an endpoint's validation, client behavior, OpenAPI, compatibility
policy, and application boundary do not become separate models as the service
grows.

---

# App architecture
Source: https://tenchi.io/architecture

Tenchi applications organize code by feature and make infrastructure and
server composition explicit.

```text
app/
  features/<feature>/
    contracts.py
    schemas.py
    ports.py
    policy.py
    routes.py
    jobs.py
    tasks.py
    tools.py
    use_cases/
    tests/
  shared/
  infra/
  server/
    context.py
    hooks.py
    webhooks.py
    routes.py
    jobs.py
    runtime.py
    preflight.py
    tasks.py
    tools.py
    mcp.py
    asgi.py
tests/
```

## Feature boundary

`contracts.py` owns HTTP declarations. `schemas.py` owns Pydantic models shared
by contracts, ports, and use cases. `ports.py` defines the interfaces the
feature needs. `policy.py` owns pure authorization rules. `routes.py` binds
contracts to use cases. `tasks.py` gives selected use cases stable operational
names. `jobs.py` declares durable background messages without importing their
consumer use cases. `tools.py` gives selected use cases stable, typed
machine-facing contracts.

Each file in `use_cases/` contains one plain async function. Unit tests sit
beside the feature and call those functions directly with memory adapters.

## Shared kernel

`app/shared/` contains application-wide errors and concepts shared by several
features, such as authenticated users. Shared code must not depend on a
feature; otherwise the dependency direction points both ways.

## Infrastructure

`app/infra/` implements feature-owned ports and exposes explicit wiring
functions. Infrastructure may depend on database drivers or external SDKs, but
never on use cases, routes, contracts, or server composition.

## Server composition

The `app/server/` package is the application root:

- `context.py` declares the frozen `AppContext`.
- `hooks.py` implements HTTP-boundary concerns such as authentication.
- `webhooks.py` binds signed-provider verification and service identity.
- `routes.py` combines feature route groups and shared error declarations.
- `jobs.py` binds feature job declarations to consumer use cases.
- `runtime.py` owns resources shared by HTTP and operational entrypoints.
- `preflight.py` declares read-only checks of the target deployment environment.
- `tasks.py` composes the application's operational task runner.
- `tools.py` composes application tools with authenticated context wiring.
- `mcp.py` optionally exposes those tools through authenticated MCP discovery
  and invocation.
- `asgi.py` creates adapters, lifecycle resources, hooks, middleware, and the
  final Starlette application.

The composition root may import anything. Everything else follows a narrower
direction.

## Dependency direction

```text
routes -> use cases -> ports -> schemas/domain
server composition -> routes
server composition -> infrastructure -> ports
server composition -> job declarations + consumer use cases
server composition -> tool bindings + authenticated context
```

Arrows point from the importing layer to the layer it depends on. Server
composition owns the complete graph; routes and infrastructure never depend
on it.

- Schemas, domain code, and ports never import infrastructure or the HTTP
  runtime.
- Use cases may import schemas, ports, policies, context types, and shared
  application code.
- Routes may import contracts and use cases, but never infrastructure.
- Job declarations may import schemas but not use cases or infrastructure;
  producers may import declarations and `job_message()`.
- Tool modules may import schemas and use cases to bind them, but never
  infrastructure or server composition. Safety annotations describe behavior;
  authorization remains in the use case.
- Infrastructure implements ports, but never imports use cases or routes.
- Server composition wires the complete graph.

Webhook verifiers live at server composition because they need secrets,
provider SDKs, and exact HTTP bytes. The verified service identity still enters
the use case through `AppContext`; provider payloads remain contract-owned
Pydantic input.

Run `uv run tenchi map --feature ` to inspect one feature and its direct
dependencies, `uv run tenchi doctor` to check these conventions alone, or
`uv run tenchi check` for the complete validation loop. Add `--json` to the map
when an agent or other tool will consume it.

  A normal `from app.server.context import AppContext` import is the clearest
  default. If an application needs to avoid that runtime import, enable
  postponed annotations with `from __future__ import annotations`, then import
  `AppContext` inside an `if TYPE_CHECKING:` block. Without postponed or quoted
  annotations, Python raises `NameError` when it defines the use case. Route
  binding ignores this app-owned context annotation while still checking every
  contract-owned boundary annotation.

```python
from __future__ import annotations

from typing import TYPE_CHECKING

from app.features.todos.schemas import CreateTodo, Todo

if TYPE_CHECKING:
    from app.server.context import AppContext

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

---

# Comparisons
Source: https://tenchi.io/comparisons

Tenchi is not a universal replacement for every Python web framework. It is a
focused choice for typed JSON APIs whose contracts, clients, and application
architecture need to remain aligned over time.

## At a glance

| Framework | Strongest fit | Main tradeoff |
| --- | --- | --- |
| FastAPI | Fast adoption, broad ecosystem, interactive API development | More built-in API ergonomics; less prescribed application architecture; typed clients generated from OpenAPI |
| Starlette | Small ASGI toolkit and low-level control | Direct control over ASGI primitives without a shared contract, client, or application model |
| Litestar | Feature-rich typed ASGI applications | More built-in capabilities and framework concepts |
| Django Ninja | Typed APIs inside the Django ecosystem | Deep Django integration and conventions rather than a framework-independent stack |
| Tenchi | Long-lived typed JSON APIs with explicit use cases and ports | More up-front structure; fewer built-in integrations and protocols |

## Choose Tenchi when

- The server and a Python client should share one executable contract.
- API compatibility must be checked in CI against a historical baseline.
- The same use cases should be exposed to AI callers through versioned,
  verifiable tool contracts.
- Business behavior should run unchanged from HTTP, workers, scripts, and
  direct tests.
- You want composition-time signature failures and explicit dependency wiring.
- A small, readable core matters more than a large extension ecosystem.

## Choose another framework when

- You need WebSockets, HTML templates, an ORM, admin UI, or background runtime
  as first-class framework features.
- You want built-in dependency injection, authentication primitives, or a
  broader integration surface.
- The API is small enough that a prescribed contract, use-case, and port
  structure would add more ceremony than value.
- You need older Python versions; Tenchi requires Python 3.12 or newer.

## Tenchi and Starlette

Tenchi exposes Starlette where that is useful. `create_app()` returns a
Starlette application, accepts Starlette middleware, and allows explicitly
declared passthrough responses for streaming, files, and redirects. Tenchi adds
contract ownership around those primitives rather than replacing the ASGI
ecosystem.

## Tenchi and FastAPI

Both frameworks use Python annotations and Pydantic at the boundary. FastAPI
optimizes for concise route declaration and ecosystem reach. Tenchi separates
the contract from the use case so the same application function is not owned
by HTTP, and it includes a contract-driven runtime client and conservative
OpenAPI compatibility analysis.

The tradeoff is visible: Tenchi asks you to name the layers. That is valuable
when the API will grow, but unnecessary ceremony for some small services.

---

# Contracts
Source: https://tenchi.io/contracts

A contract is the source of truth for one HTTP operation. It is immutable data,
separate from the use case that implements the operation.

## Declare an operation

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

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

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,
    name="create_todo",
    summary="Create a todo",
    tags=("todos",),
)
```

Pydantic's `TypeAdapter` validates every declared boundary type. You may use
Pydantic models, dataclasses, standard collections, unions, and other types
Pydantic supports; inputs are not restricted to `BaseModel` subclasses.

## Input sources

Contracts name four validated input sources:

| Option | Source | Use-case parameter |
| --- | --- | --- |
| `request=` | Request body | `request` |
| `params=` | Path parameters | `params` |
| `query=` | Query string | `query` |
| `headers=` | Request headers | `headers` |

```python
class GetTodoParams(BaseModel):
    todo_id: str

class RequestHeaders(BaseModel):
    api_version: str = Field(alias="X-API-Version")

get_todo_contract = contract(
    method="GET",
    path="/todos/{todo_id}",
    params=GetTodoParams,
    headers=RequestHeaders,
    response=Todo,
)
```

Field aliases are the wire names for path, query, and header fields and appear
the same way in OpenAPI. `route()` verifies that path placeholders match the
declared parameter schema.

## Media types

JSON is the default. Declare a different body or response representation with
`request_media_type=` and `response_media_type=`. Text declarations validate a
string; other non-JSON media types validate bytes or strings. Requests with a
missing or mismatched `Content-Type` fail with a framework-owned `415` before
decoding.

Charset-qualified text types are encoded and decoded strictly. An unsupported
declared charset fails when the contract is built rather than during a call.

## Metadata and visibility

`summary`, `description`, and `tags` feed OpenAPI. `deprecated` accepts `True`
or a timestamp; `sunset` records the removal date.

`public` defaults to `False`. Authentication hooks can use `public=True` to
exempt an operation, and OpenAPI uses the same value when security schemes are
configured. The metadata does not authenticate or authorize a request by
itself.

`webhook=True` marks a signed inbound operation. Unlike descriptive metadata,
it is enforced: `create_app()` requires a matching exact-body verifier binding.
See [Receive signed webhooks](/webhooks).

  Authentication hooks should exempt operations through
  `info.contract.public`, not URL paths or documentation tags. OpenAPI uses the
  same metadata to remove global security from public operations.

## Runtime limits

- `max_request_bytes=` overrides the application's request-body limit for one
  operation.
- `timeout=` cooperatively cancels overdue work, waits for request-scope
  cleanup, and returns a framework-owned `504`.
- `errors=` declares the application failures callers are allowed to observe.
- `responses=` declares status-dependent successful outcomes.

Invalid combinations and unsupported schemas fail when `contract()` is called.
Continue with [routes and server](/server), [responses](/responses), and
[errors](/errors).

---

# Use cases and ports
Source: https://tenchi.io/application

Tenchi keeps application behavior independent of HTTP and concrete
infrastructure.

## Define an application port

The feature owns interfaces for the capabilities it needs:

```python
from typing import Protocol

from .schemas import Todo

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

    async def get(self, todo_id: str) -> Todo | None: ...

    async def list(self) -> list[Todo]: ...
```

Nothing in this interface selects SQLAlchemy, PostgreSQL, an external service,
or a memory dictionary. Infrastructure makes that decision later.

## Carry dependencies in context

```python
from dataclasses import dataclass

from app.features.todos.ports import TodoRepository

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

The context is ordinary application data. It can also carry authenticated
identity or request-scoped ports. Frozen contexts make enrichment explicit:
an authentication hook returns `dataclasses.replace(context, user=user)`
instead of mutating shared state.

## Write a use case

```python
from app.server.context import AppContext

from ..schemas import CreateTodo, Todo

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

Use cases are plain async functions. Parameter names connect them to contract
inputs: `request`, `params`, `query`, `headers`, and `context`. A function takes
only the values it needs.

The return value is application data. HTTP headers and status selection remain
at the [route and response boundary](/responses).

## Implement an adapter

```python
from uuid import uuid4

from app.features.todos.schemas import Todo

class MemoryTodoRepository:
    def __init__(self) -> None:
        self._todos: dict[str, Todo] = {}

    async def create(self, *, title: str) -> Todo:
        todo = Todo(id=uuid4().hex, title=title, completed=False)
        self._todos[todo.id] = todo
        return todo

    async def get(self, todo_id: str) -> Todo | None:
        return self._todos.get(todo_id)

    async def list(self) -> list[Todo]:
        return list(self._todos.values())
```

Static typing verifies that the adapter satisfies `TodoRepository` when it is
returned from an explicitly annotated wiring function.

## Test without HTTP

```python
async def test_create_todo() -> None:
    repository = MemoryTodoRepository()
    context = AppContext(todos=repository)

    todo = await create_todo(CreateTodo(title="Buy milk"), context)

    assert todo.title == "Buy milk"
    assert await repository.get(todo.id) == todo
```

This is the default testing level for business behavior. Add HTTP tests for
the boundary guarantees that only the composed application can provide.

---

# Routes and server
Source: https://tenchi.io/server

Routes bind contracts to use cases. The server composes routes with concrete
dependencies and ASGI concerns.

## Bind a route

```python
from tenchi.routes import route, route_group

from .contracts import create_todo_contract
from .use_cases.create_todo import create_todo

routes = route_group(
    route(create_todo_contract, create_todo),
)
```

`route()` checks the use-case signature immediately. A missing input, wrong
parameter name, incompatible annotation, invalid header projector, or invalid
presenter is a composition error.

## Compose route groups

```python
from tenchi.routes import route_group

from app.features.notes.routes import routes as note_routes
from app.features.todos.routes import routes as todo_routes

api_routes = route_group(
    todo_routes,
    note_routes,
    prefix="/api",
)
```

Groups flatten to an immutable route collection. `errors=` declares an
application error across the group. Prefixes and shared errors create
composition, not hidden runtime behavior.

## Create the ASGI application

```python
import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from tenchi.server import create_app

from app.infra.port_wiring import ensure_schema, open_todo_repository
from app.server.context import AppContext
from app.server.routes import routes

DATABASE_PATH = os.environ.get("APP_DATABASE", "app.db")

def build_app(database_path: str = DATABASE_PATH) -> Starlette:
    @asynccontextmanager
    async def lifespan() -> AsyncIterator[str]:
        await ensure_schema(database_path)
        yield database_path

    @asynccontextmanager
    async def create_context(path: str) -> AsyncIterator[AppContext]:
        async with open_todo_repository(path) as todos:
            yield AppContext(todos=todos)

    return create_app(
        routes=routes,
        context_factory=create_context,
        lifespan=lifespan,
    )

app = build_app()
```

This matches the generated application's resource model. The lifespan prepares
the database schema once at startup. Each request enters its own repository
scope, which can commit on success, roll back on failure, and close before
Tenchi finalizes the response. `app.server.routes:routes` composes the API,
OpenAPI, Swagger UI, and health routes at the composition root.

`create_app()` returns a Starlette application and rejects duplicate or
equivalent route templates and invalid context-factory shapes during
composition. Operations on one path shape must use the same path-parameter
names. Runtime matching is independent of declaration order: literal and
constrained paths take precedence over broader parameterized paths. A `GET`
contract handles `HEAD` automatically unless a matching explicit `HEAD`
contract takes precedence.

## Lifespan and request scopes

An async context manager can create process-scoped resources once:

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan() -> AsyncIterator[Database]:
    database = await Database.connect()
    try:
        yield database
    finally:
        await database.close()

def create_context(database: Database) -> AppContext:
    return AppContext(todos=SqlTodoRepository(database))

app = create_app(
    routes=routes,
    lifespan=lifespan,
    context_factory=create_context,
)
```

The context factory may return a context directly, await one, or return an
async context manager. The final form creates request-scoped resources and
guarantees cleanup before the response is finalized.

## Hooks, middleware, and observers

- `hooks=` run after route matching and context creation, before input
  validation. They may return an enriched context and are the right boundary
  for authentication.
- `webhooks=` binds exact-body verifiers to contracts marked `webhook=True`.
  Verifiers run after ordinary hooks and body size/media checks, but before
  request parsing.
- `middleware=` accepts ordinary Starlette `Middleware` values for CORS,
  sessions, trusted hosts, compression, or app-specific ASGI behavior.
- `observers=` receive immutable `RequestOutcome` values after a matched
  request finalizes. Each outcome records its UTC `completed_at` before
  observer delivery. Observer failures are logged and cannot change the
  response.
- `use_case_observers=` receive immutable `UseCaseOutcome` values after the
  request scope closes, but only when the matched use case ran. The same
  observer contract works with `execute()`; `completed_at` records when the
  use-case call returned or raised, before later context cleanup.

  Tenchi does not wrap Starlette middleware or ASGI deployment servers. Use
  those libraries directly at the composition root and keep framework-owned
  behavior focused on contracts and use cases.

---

# Responses
Source: https://tenchi.io/responses

Tenchi validates successful response bodies and headers before they cross the
HTTP boundary. Use the simple contract fields when success has one fixed
shape; use response definitions when status or representation depends on the
result.

## One successful response

```python
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,
)

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

route(
    create_todo_contract,
    create_todo,
    response_headers=create_todo_headers,
)
```

The synchronous header projector keeps HTTP metadata out of the use case.
Tenchi checks that its return annotation and fixed scalar fields match the
contract at composition time.

## Status-dependent responses

```python
from dataclasses import dataclass
from tenchi.responses import PresentedResponse, present, response

@dataclass(frozen=True, slots=True)
class PutTodoResult:
    todo: Todo
    created: bool

created = response(Todo, status=201, description="Todo created")
existing = response(Todo, status=200, description="Todo replaced")

put_todo_contract = contract(
    method="PUT",
    path="/todos/{todo_id}",
    request=CreateTodo,
    responses=(created, existing),
)

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

put_todo_route = route(put_todo_contract, put_todo, present=present_put)
```

The use case returns domain-shaped data. A synchronous presenter selects the
declared wire response. The typed client exposes the selected definition on
`ClientResponse.definition`.

## Alternative body schemas

When one status accepts alternative top-level bodies, pass them separately so
Pyright preserves the precise union:

```python
found = response(Todo, ArchivedTodo, status=200)
```

Nested unions use ordinary Python spelling:

```python
many = response(list[Todo | ArchivedTodo], status=200)
```

## Empty and passthrough responses

Use `response(None, status=204)` for an empty result and select it with
`present(definition)`. When an empty definition also uses passthrough, its
concrete Starlette response must have a materialized empty body.

For streaming, files, or redirects, declare `passthrough=True` and present a
Starlette `Response`. Tenchi preserves it while validating the contract-owned
status, media type parameters, and declared headers.

  Passthrough is not an escape hatch around the contract. A streaming response
  declares its body type, and its status, content type, and headers must match
  the selected response definition.

---

# Errors
Source: https://tenchi.io/errors

Application errors are stable public outcomes. Define them once, raise them
from use cases, and declare which contracts may expose them.

## Define and declare an error

```python
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,),
)
```

Raise the definition with request-specific safe details:

```python
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
```

## Return declared error headers

List any application-owned response headers on the definition, then provide
their values when raising the error:

```python
from tenchi.rate_limits import RATE_LIMITED, enforce_rate_limit

async def create_todo(request: CreateTodo, context: AppContext) -> Todo:
    user = require_user(context.user)
    await enforce_rate_limit(
        context.rate_limits,
        namespace="todos.create",
        scope=user.id,
        limit=10,
        window_seconds=60,
    )
    return await context.todos.create(title=request.title)
```

Tenchi rejects undeclared or unsafe header values, documents declared headers
in OpenAPI, and validates them in the typed client. Declare `RATE_LIMITED` on
the contract or a containing route group like any other application error. See
[Limit application operations](/rate-limits) for the store and scoping model.

## The honesty rule

An `AppError` maps to its declared status only when the matched contract or a
containing route group declares that definition. An undeclared application
error becomes a framework-owned `500` instead of leaking an undocumented
behavior.

The same rule applies to errors raised by authentication hooks. Every private
route that can receive an authentication failure must declare it directly or
through a route group.

## Error envelope

Application and framework errors use one flat shape:

```json
{
  "code": "TODO_NOT_FOUND",
  "message": "Todo not found",
  "details": { "todo_id": "abc123" },
  "request_id": "01J..."
}
```

Every error response carries `x-tenchi-error-source: app | framework`.
Framework failures use stable codes for validation, media types, body limits,
timeouts, missing routes, and internal contract violations.

## Client symmetry

The typed client raises declared `AppError` values. An undocumented status,
invalid envelope, wrong error source, mismatched media type, or invalid success
body raises `UnexpectedResponseError`.

Pass shared group errors to `Client(errors=...)` when the server declares them
at a route-group boundary. This keeps client and server error semantics
symmetric.

  Treat `message`, `details`, and declared headers as API data. Do not include
  exception text, SQL, tokens, or internal identifiers that callers should not
  receive.

---

# Typed client
Source: https://tenchi.io/client

`Client` is an async httpx client driven by the same contracts as the server.
It serializes validated inputs and refuses responses that violate the declared
wire shape.

## Call a contract

```python
from tenchi.client import Client

async with Client(base_url="https://api.example.com") as client:
    todo = await client.call(
        create_todo_contract,
        request=CreateTodo(title="Buy milk"),
    )
```

`call()` returns only the validated response body. Input arguments mirror the
contract: `request=`, `params=`, `query=`, and `headers=`.

## Inspect the HTTP response

```python
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
    definition = created.definition
```

`call_with_response()` adds validated successful headers, the underlying
`httpx.Response`, and the selected status-dependent response definition.

## Configure transport and headers

```python
import httpx

transport = httpx.AsyncHTTPTransport()

async with Client(
    base_url="https://api.example.com",
    headers={"Authorization": f"Bearer {token}"},
    transport=transport,
) as client:
    ...
```

Tenchi exposes httpx rather than wrapping its transport model. You may instead
pass an existing `httpx.AsyncClient` with `http=`; Tenchi does not close a
client it does not own.

## Retry a logical call

Calls make one HTTP attempt by default. Pass an explicit policy when a
dependency and operation are safe to retry:

```python
from tenchi.retries import retry_policy

dependency_retry = retry_policy(
    max_attempts=3,
    retry_on=("TEMPORARILY_UNAVAILABLE",),
    base_delay_seconds=0.2,
    max_delay_seconds=2,
    total_timeout_seconds=5,
)

todo = await client.call(
    get_todo_contract,
    params=GetTodoParams(todo_id=todo_id),
    retry=dependency_retry,
)
```

Transport failures are retryable by default within the policy. Declared
application errors are retried only when their stable code appears in
`retry_on`. Tenchi uses exponential backoff with bounded jitter and waits at
least as long as a valid `Retry-After` delay or HTTP date carried by the
declared error. The total timeout covers attempts and backoff; exceeding it
raises `RetryTimeoutError`. Caller cancellation always remains cancellation.

Unexpected statuses, invalid media types, invalid error envelopes, and invalid
success bodies or headers are never retried. They indicate contract drift or
bad data, not a transient dependency.

GET, HEAD, OPTIONS, and TRACE may use a policy directly. Retrying POST, PUT,
PATCH, or DELETE requires `allow_unsafe_methods=True`:

```python
command_retry = retry_policy(
    max_attempts=3,
    total_timeout_seconds=5,
    allow_unsafe_methods=True,
)

created = await client.call(
    create_todo_contract,
    headers=CreateTodoHeaders(idempotency_key=key),
    request=request,
    retry=command_retry,
)
```

That flag records an application decision; it does not make the operation
safe. Reuse one caller-generated [idempotency key](/idempotency) across every
attempt, or make the remote operation idempotent by design.

## Response enforcement

The client validates:

- the selected successful status,
- response `Content-Type` and charset,
- the response body,
- declared successful headers,
- declared application error envelopes and sources.

Unexpected statuses, media types, and error envelopes raise
`UnexpectedResponseError` with the contract name, status, body, and reason.
Pydantic raises `ValidationError` when a declared success body or header value
does not validate.

## Observe outbound calls

Pass synchronous or asynchronous observers when you construct the client:

```python
import logging

from tenchi.client import Client, ClientOutcome

logger = logging.getLogger("app.outbound")

def observe_client(outcome: ClientOutcome) -> None:
    logger.info(
        "client.complete",
        extra={
            "operation": outcome.contract.name,
            "status": outcome.status,
            "status_code": outcome.status_code,
            "duration_seconds": outcome.duration_seconds,
            "completed_at": outcome.completed_at.isoformat(),
            "error_code": outcome.error_code,
            "attempts": outcome.attempts,
        },
    )

async with Client(
    base_url="https://api.example.com",
    observers=[observe_client],
) as client:
    todo = await client.call(
        create_todo_contract,
        request=CreateTodo(title="Buy milk"),
    )
```

Each call produces one immutable `ClientOutcome` before it returns or raises:

| Field | Meaning |
| --- | --- |
| `contract` | The contract used for the call; use its stable `name` for the operation |
| `status` | The outcome classification described below |
| `status_code` | The HTTP status when a response arrived; otherwise `None` |
| `duration_seconds` | Time spent preparing input, waiting for transport and retry backoff, and validating responses; observer work is excluded |
| `completed_at` | UTC completion time captured before logical-call observers run |
| `error_code` | The declared `AppError` code for `app_error`; otherwise `None` |
| `attempts` | Number of transport attempts made for this logical call |

The status distinguishes the part of the outbound boundary that failed:

| Status | Meaning |
| --- | --- |
| `succeeded` | The response matched a declared successful outcome |
| `app_error` | The remote application returned a declared application error |
| `unexpected_response` | A response arrived but its status, media type, body, headers, or error envelope violated the contract |
| `transport_error` | httpx could not complete the transport operation |
| `timed_out` | The explicit retry policy exhausted its total timeout |
| `failed` | No response arrived and the failure was not an httpx transport error or cancellation, such as invalid local input or contract configuration |
| `cancelled` | The client call was cancelled |

Outcomes never include input values, request URLs or headers, response bodies
or headers, or exception objects. This keeps the default observer surface safe
for metrics and structured operational logs. If you instrument the underlying
httpx transport separately, apply your own URL, header, and payload redaction.

Observers run in declaration order and may be sync or async. Their failures are
logged and isolated from later observers and from the value or exception the
caller receives. Keep them fast because the call waits for the observer chain.

To see individual attempts, pass `attempt_observers=` and accept
`ClientAttemptOutcome`. It reports the attempt number, maximum attempts,
classification, status and error code when available, whether the policy
scheduled another attempt, its selected delay, and the attempt's UTC completion
time. A deadline can still expire during that delay. One logical
`ClientOutcome` is emitted after the retry sequence:

```python
from tenchi.client import ClientAttemptOutcome

def observe_attempt(outcome: ClientAttemptOutcome) -> None:
    logger.info(
        "dependency_attempt",
        extra={
            "operation": outcome.contract.name,
            "status": outcome.status,
            "will_retry": outcome.will_retry,
            "completed_at": outcome.completed_at.isoformat(),
        },
    )
```

Attempt outcomes carry the same payload-safety guarantee as logical outcomes.
Use logical outcomes for availability and latency service-level indicators;
use attempt outcomes for retry pressure and dependency diagnostics.

## In-process use

Tests should normally use `tenchi.testing.open_client(app)`. It supplies an
ASGI transport, runs the application lifespan, and returns the same `Client`
API used against a deployed server. Pass `observers=` and
`attempt_observers=` to assert outbound outcomes in a test.

---

# Pagination
Source: https://tenchi.io/pagination

Tenchi provides one small limit-and-offset vocabulary that remains typed across
the contract, use case, client, and OpenAPI document.

## Declare a query and page

```python
from tenchi.contracts import contract
from tenchi.pagination import Page, PageQuery

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

list_todos_contract = contract(
    method="GET",
    path="/todos",
    query=ListTodosQuery,
    response=Page[Todo],
)
```

`PageQuery` declares `limit` and `offset` with boundary validation. Subclass it
to add filters or sorting fields owned by the operation.

`Page[Todo]` serializes one stable envelope:

```json
{
  "items": [],
  "total": 0,
  "limit": 20,
  "offset": 0
}
```

## Build the result

```python
from tenchi.pagination import Page, page

async def list_todos(
    query: ListTodosQuery,
    context: AppContext,
) -> Page[Todo]:
    items, total = await context.todos.list_page(
        limit=query.limit,
        offset=query.offset,
        completed=query.completed,
    )
    return page(items, total=total, query=query)
```

`page()` copies the validated limit and offset into the response, keeping the
repository result and public envelope aligned.

## Keep pagination in the port

Memory tests may slice a list, but production repositories should apply limit,
offset, and filters in the database or remote service. Return the current page
plus the total matching count through the feature-owned port; the use case
turns those values into the public `Page`.

The typed client returns `Page[Todo]`, and OpenAPI includes both the query
constraints and the generic item schema without another declaration.

---

# Authentication and authorization
Source: https://tenchi.io/authentication

Authentication belongs at the HTTP boundary. Business authorization belongs
in use cases and pure policy functions. Keeping them separate lets the same
application rules run from HTTP, workers, scripts, and direct tests.

## Carry optional identity

```python
from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class AppContext:
    todos: TodoRepository
    user: User | None = None
```

The context begins without verified identity. A hook authenticates the request
and returns an enriched copy.

## Authenticate in a hook

```python
from dataclasses import replace

from tenchi.errors import AppError
from tenchi.server import Hook, RequestInfo

def create_bearer_hook(tokens: TokenDirectory) -> Hook:
    async def authenticate(
        info: RequestInfo,
        context: AppContext,
    ) -> AppContext | None:
        if info.contract.public:
            return None

        scheme, _, token = info.headers.get("authorization", "").partition(" ")
        if scheme.lower() != "bearer" or not token:
            raise AppError(unauthorized)

        user = await tokens.lookup(token)
        if user is None:
            raise AppError(unauthorized)
        return replace(context, user=user)

    return authenticate
```

Register the hook at composition:

```python
app = create_app(
    routes=routes,
    context_factory=create_context,
    hooks=(create_bearer_hook(tokens),),
)
```

Authentication hooks run before request input validation, so an unauthenticated
caller cannot use validation responses to inspect an operation the hook
protects.

Signed provider callbacks need the exact request bytes, not only headers. Use
a [webhook verifier binding](/webhooks) for those endpoints. It runs after
ordinary hooks and request size/media checks, before contract input validation,
and may attach a service identity to the same context model.

## Declare authentication failures

The server's [error honesty rule](/errors) applies to hooks. Declare the
authentication error across the routes protected by the hook:

```python
private_routes = route_group(
    project_routes,
    task_routes,
    errors=(unauthorized,),
)
```

Health and OpenAPI routes are public by default. Pass `public=False` if the
application should protect them.

## Authorize in the use case

```python
async def get_project(
    params: GetProjectParams,
    context: AppContext,
) -> Project:
    user = require_user(context.user)
    project = await context.projects.get(params.project_id)
    return ensure_can_view_project(user, project, project_id=params.project_id)
```

The use case still asserts identity through an app-owned helper. This matters
when a worker or test invokes it without the HTTP hook. It fetches the subject
through a port, then delegates the authorization decision to the subject's
policy module.

The policy keeps both the reusable ability and its declared failure mapping in
one I/O-free place:

```python
from app.shared.errors import project_not_found
from tenchi.errors import AppError

def can_view_project(user: User, project: Project | None) -> bool:
    if project is None:
        return False
    return project.owner_id == user.id or user.id in project.member_ids

def ensure_can_view_project(
    user: User,
    project: Project | None,
    *,
    project_id: str,
) -> Project:
    if project is None or not can_view_project(user, project):
        raise AppError(project_not_found, details={"project_id": project_id})
    return project
```

`can_view_project()` can be reused by other features. The `ensure_*` helper
turns the same decision into the operation's public error semantics. On reads,
missing and unviewable subjects both become not-found so callers cannot probe
identifiers.

  Setting `public=True` only gives hooks and OpenAPI one consistent exemption
  signal. The hook must still inspect it, and use cases remain responsible for
  business authorization.

For webhook contracts, `public=True` exempts only the ordinary authentication
hook. `webhook=True` separately requires an exact-body verifier at application
composition.

---

# Workers and scripts
Source: https://tenchi.io/execution

HTTP is one caller of an application use case. `execute()` applies Tenchi's
request validation and context-scoping rules from jobs, command-line scripts,
message consumers, or tests without inventing a second contract.

## Execute validated input

```python
from tenchi.execution import execute

async def handle_job(payload: bytes, context: AppContext) -> Todo:
    return await execute(
        create_todo,
        request_json=payload,
        context=context,
    )
```

`request_json=` validates serialized input against the use case's annotated
`request` parameter. Use `request=` for an already decoded Python value.

`execute()` deliberately supplies only `request` and `context`. A use case that
requires HTTP-specific `params`, `query`, or `headers` is not portable to this
entrypoint and fails with `ExecutionError`.

## Open a context explicitly

```python
from tenchi.execution import open_context

async with open_context(create_request_context()) as context:
    result = await execute(rebuild_index, context=context)
```

`open_context()` accepts a direct value, an awaitable, or an async context
manager. This mirrors the server's context behavior, including cleanup for
transactional or request-scoped resources.

## Observe the use case

Pass the same observer to HTTP composition and direct execution:

```python
import logging

from tenchi.execution import UseCaseOutcome, execute

logger = logging.getLogger("app.operations")

def observe_use_case(outcome: UseCaseOutcome) -> None:
    operation = str(
        getattr(outcome.use_case, "__name__", type(outcome.use_case).__name__)
    )
    logger.info(
        "use_case.complete",
        extra={
            "operation": operation,
            "entrypoint": outcome.entrypoint,
            "status": outcome.status,
            "duration_seconds": outcome.duration_seconds,
            "completed_at": outcome.completed_at.isoformat(),
            "error_code": outcome.error_code,
        },
    )

result = await execute(
    rebuild_index,
    context=create_request_context,
    use_case_observers=(observe_use_case,),
)
```

`create_app(use_case_observers=(observe_use_case,))` reports the HTTP path
through the same `UseCaseOutcome` shape. The observer runs after context
cleanup, while `duration_seconds` measures only the use-case call and
`completed_at` records when that call returned or raised.

No outcome is emitted when validation, context acquisition, or a boundary hook
fails before the use case starts. Observer failures are logged and do not
replace the use case's result or exception. Keep observers fast or hand work to
a non-blocking telemetry pipeline because `execute()` waits for them before it
returns.

## Keep transport concerns outside

The message consumer acknowledges, retries, or dead-letters work. The use case
owns application behavior. Convert message metadata into application input
before calling `execute()` rather than passing a queue SDK object into the
application layer.

Authorization remains in the use case. When a job acts for a user, construct a
context with verified actor identity and let the same policy checks run.

For durable delivery, idempotent commands, and retry classification, continue
with [Background jobs](/jobs) and [Retries and background
work](/reliability). Use `execute()` directly when the caller already owns the
function identity and no durable name or producer contract is needed. For
operator-invoked backfills, repairs, replays, and maintenance, use
[Operational tasks](/tasks)
to add stable discovery plus result validation around the same use cases.

---

# Background jobs
Source: https://tenchi.io/jobs

Tenchi validates the message boundary between a producer and consumer without
becoming a queue. Your infrastructure still owns persistence, claiming,
acknowledgement, retries, backoff, concurrency, and dead letters.

## Declare the message

Keep the stable name and payload type with the feature that owns the event:

```python
# app/features/projects/jobs.py
from tenchi.jobs import job

from .schemas import MemberAdded

member_added_job = job(
    "projects.member_added",
    request=MemberAdded,
    result=None,
    description="Notify a user after project membership is committed.",
)
```

Changing a job name or payload can strand messages already stored in a queue.
Treat both as durable wire contracts. Introduce a new name when a consumer
cannot safely read both payload versions.

## Validate before enqueueing

Build a `JobMessage` before handing data to an outbox or queue port:

```python
from tenchi.jobs import job_message

message = job_message(
    member_added_job,
    MemberAdded(
        project_id=project.id,
        project_name=project.name,
        user_id=user_id,
    ),
)
await context.outbox.enqueue(
    job=message.name,
    payload_json=message.payload_json,
)
```

`job_message()` validates Python input, emits compact JSON bytes, and confirms
that the declared consumer type can read those bytes back. A useful queue port
therefore stays transport-shaped:

```python
from typing import Protocol

class Outbox(Protocol):
    async def enqueue(self, *, job: str, payload_json: bytes) -> None: ...
```

The job payload carries application facts. Delivery metadata such as message
ids, trace ids, attempt counts, scheduled time, and queue partition belongs in
your infrastructure envelope, not in every business payload.

## Bind the consumer

Bind declarations to plain async use cases at the composition root:

```python
# app/server/jobs.py
from app.features.projects.jobs import member_added_job
from app.features.projects.use_cases.notify_member_added import (
    notify_member_added,
)
from tenchi.jobs import create_job_dispatcher, job_group, job_handler

jobs = job_group(
    job_handler(member_added_job, notify_member_added),
)
dispatcher = create_job_dispatcher(jobs=jobs)
```

The handler must accept `request` and `context`. Its request and return
annotations must exactly match the job declaration, so bad wiring fails when
the application imports:

```python
async def notify_member_added(
    request: MemberAdded,
    context: AppContext,
) -> None:
    await context.notifications.record(
        user_id=request.user_id,
        message=f"You joined {request.project_name}",
    )
```

`job_group()` rejects duplicate names. Put every registered group in
`app/server/jobs.py`; `tenchi map` then shows job nodes and their handler
bindings.

## Dispatch one delivery

After your worker claims a message and creates its unit-of-work context,
dispatch the raw stored JSON:

```python
await dispatcher.dispatch(
    entry.job,
    payload_json=entry.payload_json,
    context=context,
)
```

The dispatcher:

1. rejects unknown names before opening the supplied context;
2. validates JSON before the handler runs;
3. invokes the handler through the shared use-case observer boundary;
4. validates the result before the context exits successfully.

If you pass an async context manager or factory, result validation happens
before its successful exit, so an invalid result can still roll back the unit
of work. Passing a ready context leaves commit and rollback with the caller.

| Failure | Typical worker decision |
| --- | --- |
| `JobNotFoundError` | Dead-letter; deploying the correct consumer may make a later replay possible |
| Pydantic `ValidationError` | Dead-letter; the stored payload does not match the declared request |
| `JobResultError` | Roll back and dead-letter; the handler violates its result contract |
| `AppError` | Decide from the stable application error code |
| Cancellation | Release or roll back the claim and stop promptly |
| Dependency or transport failure | Roll back, apply bounded backoff, and retry |

  Commit application writes and acknowledgement together when your queue or
  outbox supports it. A worker can still crash after an external service
  accepts work but before acknowledgement commits, so consumers must be
  idempotent or use a downstream idempotency key.

## Observe handlers

Pass `use_case_observers=` to `create_job_dispatcher()`. Each
`UseCaseOutcome.entrypoint` is `"job"` and contains only the use-case identity,
status, duration, UTC completion time, and stable application error code. Queue
latency, attempt number, message id, and dead-letter state remain worker
telemetry because the dispatcher never owns them.

The taskboard example demonstrates a SQLite transactional outbox, typed
producer message, registered dispatcher, rollback before dead-lettering, and
retry ownership in the worker. Continue with [Retries and background
work](/reliability) for the complete transaction and failure-classification
pattern.

---

# Operational tasks
Source: https://tenchi.io/tasks

Operational tasks give backfills, repairs, replays, and maintenance commands a
named, validated entrypoint. They run ordinary async use cases with the same
application lifespan, scoped context, and use-case observers as the HTTP
application.

Tenchi does not schedule tasks or turn them into a queue. Your deployment
decides who may invoke them, when they run, and whether a failed operation
should be retried.

## Declare the input, result, and use case

Use Pydantic models for inputs and results. A safe maintenance command normally
supports a dry run:

```python
from pydantic import BaseModel

from app.server.context import AppContext

class RepairMembersInput(BaseModel):
    dry_run: bool = True

class RepairMembersResult(BaseModel):
    scanned: int
    repaired: int
    dry_run: bool

async def repair_project_members(
    request: RepairMembersInput,
    context: AppContext,
) -> RepairMembersResult:
    scanned, repaired = await context.projects.repair_invalid_member_ids(
        dry_run=request.dry_run,
    )
    return RepairMembersResult(
        scanned=scanned,
        repaired=repaired,
        dry_run=request.dry_run,
    )
```

Keep this function in the feature's `use_cases/` directory and test it directly
with memory adapters. It remains application behavior rather than CLI code.

## Give the task a stable name

Bind the use case in `app/features/projects/tasks.py`:

```python
from tenchi.tasks import task, task_group

from .use_cases.repair_project_members import repair_project_members

repair_project_members_task = task(
    "projects.repair_members",
    repair_project_members,
    description="Replace malformed project member lists with an empty list.",
)

tasks = task_group(repair_project_members_task)
```

Task names use dotted `snake_case`. Treat them as operator-facing API names:
scripts and agents may keep them long after the Python function moves.
`task()` checks the use-case signature and builds the input and result
validators when the module is imported.

## Compose the runner

HTTP and tasks should share application resource wiring. A small
`app/server/runtime.py` can own `create_context` and the lifespan factory. Then
compose tasks in `app/server/tasks.py`:

```python
from app.features.projects.tasks import tasks as project_tasks
from app.server.observability import observe_use_case
from app.server.runtime import DATABASE_PATH, create_context, create_lifespan
from tenchi.tasks import create_task_runner, task_group

tasks = task_group(project_tasks)

runner = create_task_runner(
    tasks=tasks,
    context_factory=create_context,
    lifespan=create_lifespan(DATABASE_PATH),
    use_case_observers=(observe_use_case,),
)
```

The default CLI target is `app.server.tasks:runner`. Use `--tasks
module:attribute` when your composition root lives elsewhere.

## Discover and run tasks

List names and their JSON Schemas before invoking anything:

```shell
uv run tenchi task list
uv run tenchi task list --json
```

Run the dry run first, inspect its result, then opt into the write:

```shell
uv run tenchi task run projects.repair_members \
  --input '{"dry_run": true}'

uv run tenchi task run projects.repair_members \
  --input '{"dry_run": false}' \
  --json
```

Input is validated before the lifespan or context opens. Output is validated
before the scoped context commits, so an invalid result rolls back
transactional work. Application errors, invalid input, invalid results, and
unexpected failures have distinct codes in JSON output. Task input is not
echoed into the result.

Cancellation propagates through the use case and both cleanup scopes. A
cancelled process or MCP call can therefore roll back the same way as another
failure, provided the database adapter follows the documented context-manager
pattern.

Use-case observers receive `entrypoint="task"` so task telemetry stays
distinguishable from HTTP and direct `execute()` calls.

  `tenchi task run` uses the process's credentials and application context. Run
  it only from an operator-controlled environment. Authentication hooks are an
  HTTP concern and do not run here; keep business authorization in reusable use
  cases when a task acts for a user.

## Let an agent discover tasks

The MCP server always exposes the read-only `task_list` tool. Task execution is
disabled unless the server starts with an explicit capability:

```shell
uv run tenchi mcp --allow-task-runs
```

That flag adds the state-changing `task_run` tool. Give it only to an agent and
environment that are allowed to perform operational writes. The tool returns
the same versioned result shape as `tenchi task run --json`.

`tenchi map` also includes task nodes and their bindings to use cases. This
lets an operator or coding agent see the behavior and dependencies behind a
task before running or changing it.

For recurring or event-driven work, keep the consumer's acknowledgement,
retry, and dead-letter policy around [`execute()`](/execution). See [Retries
and background work](/reliability) for that boundary.

---

# Application tools
Source: https://tenchi.io/tools

Application tools give an AI runtime or another machine caller a stable name,
validated input and output, declared errors, and safety metadata for an
existing use case. The tool boundary does not choose a model provider, agent
loop, or transport.

A `ToolRunner` owns one application lifespan and scoped context per call. It
validates input before opening either resource and validates the result before
the context commits.

## Declare and bind a tool

Keep the declaration and its binding with the feature:

```python
# app/features/projects/tools.py
from app.shared.errors import unauthorized
from tenchi.tools import tool, tool_group, tool_handler

from .schemas import Project
from .use_cases.list_projects import list_projects

search_projects_tool = tool(
    "projects.search",
    result=list[Project],
    description="List projects owned by the authenticated user.",
    errors=(unauthorized,),
    read_only=True,
    open_world=False,
)

tools = tool_group(
    tool_handler(search_projects_tool, list_projects),
)
```

The use case remains an ordinary async function:

```python
async def list_projects(context: AppContext) -> list[Project]:
    owner = require_owner_scope(context.user)
    return await context.projects.list_owned_by(owner)
```

`tool_handler()` checks the function when the module imports. A tool with a
request requires exactly annotated `request` and `context` parameters; a tool
without a request requires only `context`. The return annotation must exactly
match the declared result.

The same use case may also be bound to an HTTP route. Authentication and
presentation remain entrypoint concerns while authorization stays inside the
shared use case.

## Compose an authenticated runner

Combine feature groups at server composition, after the caller's identity has
been authenticated:

```python
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from dataclasses import replace

from app.features.projects.tools import tools as project_tools
from app.server.context import AppContext
from app.server.runtime import create_context, create_lifespan
from app.shared.users import User
from tenchi.tools import ToolRunner, create_tool_runner, tool_group

tools = tool_group(project_tools)

def create_user_tool_runner(
    *,
    database_path: str,
    user: User,
) -> ToolRunner:
    @asynccontextmanager
    async def tool_context(state: str) -> AsyncGenerator[AppContext]:
        async with create_context(state) as context:
            yield replace(context, user=user)

    return create_tool_runner(
        tools=tools,
        context_factory=tool_context,
        lifespan=create_lifespan(database_path),
    )
```

The identity is application-owned context, not model-supplied tool input.
Use cases still call `require_user()` and pure policies because the same
behavior may run through HTTP, a job, a test, or another future entrypoint.

  `read_only`, `destructive`, `idempotent`, and `open_world` describe expected
  behavior to a trusted adapter or client. They never replace authentication,
  application policies, idempotency, quotas, or an approval decision.

## Call a tool

Pass Python data or raw JSON:

```python
result = await runner.call(
    "projects.search",
)

moved = await runner.call(
    "tasks.move",
    input_json="""{
      "task_id": "task_123",
      "status": "done",
      "expected_version": 4,
      "idempotency_key": "agent-run_456"
    }""",
)
```

Input validation happens before lifespan or context acquisition. Output
validation happens inside the context scope, so a result-contract violation
can roll back transactional writes. Cancellation passes through the use case
and both cleanup scopes.

Pass `use_case_observers=` to `create_tool_runner()` to receive the normal
payload-safe `UseCaseOutcome` with `entrypoint="tool"`.

## Declare caller-visible errors

List every `ErrorDef` that a caller may receive:

```python
move_task_tool = tool(
    "tasks.move",
    request=MoveTaskInput,
    result=Task,
    description="Move one visible task to another workflow status.",
    errors=(
        unauthorized,
        task_not_found,
        forbidden,
        precondition_failed,
        IDEMPOTENCY_CONFLICT,
        IDEMPOTENCY_IN_PROGRESS,
        RATE_LIMITED,
    ),
    destructive=True,
    idempotent=True,
    open_world=False,
)
```

| Failure | Runner behavior |
| --- | --- |
| Invalid input | Raises Pydantic `ValidationError` before resources open |
| Unknown name | Raises `ToolNotFoundError` |
| Declared `AppError` | Preserves the declared application error |
| Undeclared `AppError` | Raises generic `ToolInvocationError` with the original error as its cause |
| Unexpected exception | Raises the same generic `ToolInvocationError` |
| Invalid result | Raises `ToolResultError` before context commit |
| Cancellation | Propagates cancellation through cleanup |

The generic invocation error contains the tool name but no exception message,
details, request, or result.

## Inspect the portable manifest

Use `tool_manifest()` when an adapter or a test needs deterministic discovery:

```python
from tenchi.tools import tool_manifest

manifest = tool_manifest(tools)
```

The versioned manifest sorts tools and errors by stable name. Each entry
contains the description, input and output JSON Schemas, declared error codes
and messages, and all four safety annotations. A no-input tool receives an
empty object input schema; callers may either omit input or pass `{}`.

`tool()` validates, converts to JSON, and retains both schemas when the
declaration is built. Invalid custom schemas therefore fail during composition,
and repeated manifest calls return the same schema.

`TOOL_MANIFEST_VERSION` identifies the portable manifest contract. Tenchi
uses a new version for breaking changes rather than silently changing an
adapter's assumptions.

The manifest describes registered application tools only. It does not expose
Python functions, context values, credentials, request payloads, or results.

## Protect the contract in version control

Write the composed manifest from the application root:

```shell
uv run tenchi tools --write tools.json
```

`tenchi check` compares the generated manifest with `tools.json` so an
unrecorded contract change fails locally and in CI. Before accepting a changed
snapshot, classify it against the current file:

```shell
uv run tenchi tools --diff tools.json
uv run tenchi tools --write tools.json
```

For pull requests, compare with a historical baseline even when the branch
also updates `tools.json`:

```shell
uv run tenchi tools \
  --diff-ref origin/main \
  --snapshot tools.json
```

The compatibility report treats these changes directionally:

- removing or renaming a tool is breaking;
- narrowing accepted input is breaking;
- widening possible output is breaking;
- adding a caller-visible application error is breaking;
- changing a safety annotation in the less-safe direction is breaking;
- adding a tool or making a contract provably safer is additive;
- descriptions and error messages are metadata.

Breaking and unknown changes exit non-zero. Use `--diff-format json` when an
agent or CI system needs the versioned report. `tenchi tools --json` returns
the current registered manifest in a versioned result without writing a file.

`tenchi map` includes tool nodes, their feature ownership, safety details,
registration state, and exact bindings to use cases. This lets a coding agent
find both the contract and the behavior it exposes before editing either one.

## Connect a transport

Use the manifest for discovery and call `ToolRunner.call()` from the SDK's tool
callback. The adapter remains responsible for:

- authenticating the caller and constructing the appropriate app context;
- deciding which registered tools that caller may discover;
- applying approval policy before state-changing calls;
- mapping declared errors into the transport's error shape;
- propagating cancellation and trace context.

For MCP clients, [`create_tool_mcp_server()`](/tool-mcp) maps schemas and
failures, propagates cancellation, and invokes callbacks for authentication,
visibility, and approval policy.

`tenchi mcp` is a different server: it lets a coding agent inspect and validate
a Tenchi repository. It does not expose the application's `ToolGroup`.

---

# Serve tools over MCP
Source: https://tenchi.io/tool-mcp

Use `tenchi.mcp` to expose a registered `ToolGroup` to Model Context Protocol
(MCP) clients. The adapter publishes each tool's JSON Schemas and safety
annotations, authenticates discovery and invocation, and executes calls through
the caller's `ToolRunner`.

  `tenchi.mcp` serves the application tools you declared. The `tenchi mcp` CLI
  command is a separate development server that lets coding agents inspect and
  validate a Tenchi repository.

## Install the MCP integration

Add the optional dependency to the environment that will run the server:

```shell
uv add "tenchi[mcp]"
```

Keep it as a development dependency only when you use the coding-agent server
and do not expose application tools at runtime.

## Authenticate and create the server

Authentication is application-owned. The callback receives the MCP request id
and, for HTTP transports, the underlying Starlette `Request`. Resolve identity
from that request, then build a runner whose context carries the authenticated
principal:

```python
# app/server/mcp.py
from app.infra.tokens import token_directory
from app.server.tools import create_user_tool_runner, tools
from app.shared.users import User
from tenchi.mcp import McpRequest, create_tool_mcp_server

async def authenticate(request: McpRequest) -> User:
    transport = request.transport_request
    if transport is None:
        raise PermissionError("This server requires HTTP authentication.")

    scheme, _, token = transport.headers.get("authorization", "").partition(" ")
    if scheme.lower() != "bearer" or not token:
        raise PermissionError("A bearer token is required.")

    user = await token_directory.lookup(token)
    if user is None:
        raise PermissionError("The bearer token is invalid.")
    return user

mcp = create_tool_mcp_server(
    tools=tools,
    authenticate=authenticate,
    runner_factory=lambda user: create_user_tool_runner(user=user),
    name="Acme backend",
    instructions="Search records freely. Request approval before changing them.",
)

app = mcp.streamable_http_app()
```

Serve the returned ASGI application with its lifespan enabled:

```shell
uv run uvicorn app.server.mcp:app
```

The default Streamable HTTP endpoint is `/mcp`. Tenchi uses the MCP SDK's
stateless HTTP mode, so calls do not depend on sticky routing to an in-memory
MCP session. Application resources still follow the lifespan and context
wiring in each caller's `ToolRunner`.

The MCP SDK's default DNS-rebinding policy accepts loopback hosts for local
development. Configure the public host—and any browser origins you
allow—before deployment:

```python
from mcp.server.transport_security import TransportSecuritySettings

mcp = create_tool_mcp_server(
    tools=tools,
    authenticate=authenticate,
    runner_factory=lambda user: create_user_tool_runner(user=user),
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=["mcp.example.com"],
        allowed_origins=["https://console.example.com"],
    ),
)

app = mcp.streamable_http_app()
```

Requests with another `Host` or `Origin` are rejected before authentication.
Use your deployed host names rather than the placeholders above.

To mount `app` inside another Starlette application, explicitly enter the MCP
app's lifespan from the parent lifespan; Starlette does not run a mounted
sub-application's lifespan automatically:

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.routing import Mount

mcp_app = mcp.streamable_http_app()

@asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
    del app
    async with mcp.session_manager.run():
        yield

app = Starlette(
    routes=[Mount("/tools", app=mcp_app)],
    lifespan=lifespan,
)
```

This mount serves MCP at `/tools/mcp`. Deployment remains responsible for TLS,
trusted hosts, origin policy, edge rate limits, and any network-level access
control.

Authentication runs again for every `tools/list` and `tools/call` request.
Authentication failures become generic MCP errors; callback exception messages
are not returned to the client.

## Limit discovery per caller

Use `allow_tool` when different principals should discover different tool
names. In this example, `allowed_tools` is application-owned identity data:

```python
from typing import Any

from app.shared.users import User
from tenchi.tools import Tool

def allow_tool(user: User, declaration: Tool[Any, Any]) -> bool:
    return declaration.name in user.allowed_tools
```

Pass the callback to `create_tool_mcp_server(allow_tool=allow_tool)`. The
adapter applies it during discovery and rechecks it before every call. A hidden
tool behaves like an unknown tool.

Discovery filtering is not business authorization. Use cases still assert
identity and enforce policies because the same behavior may run through HTTP,
a job, a script, or a direct test.

## Require approval for destructive tools

A tool declared with `destructive=True` cannot run until an `approve` callback
accepts that principal, tool declaration, and normalized JSON input. Tenchi
validates the input once, gives the callback a JSON copy with serialization
aliases applied, and passes the validated Python value to the use case. Changes
to the callback's copy cannot change the invocation. Here, `approvals` is an
application-owned durable store that consumes one approval for the exact call:

```python
from typing import Any

from app.shared.users import User
from tenchi.tools import Tool

async def approve(
    user: User,
    declaration: Tool[Any, Any],
    input_value: object,
) -> bool:
    return await approvals.consume(
        user_id=user.id,
        tool_name=declaration.name,
        input_value=input_value,
    )
```

Without an approval callback, the call returns `approval_required`. A callback
that returns `False` produces `approval_denied`. Both results are structured,
successful MCP responses so a client can obtain approval out of band and retry.
The adapter validates input and obtains approval before opening the tool
runner's lifespan or context.

  Approval answers whether this caller may attempt this exact destructive
  action now. The use case must still enforce ownership, roles, quotas,
  idempotency, and domain policy.

## Understand MCP input and output

MCP requires object-shaped tool arguments. Tenchi maps declarations
predictably:

| Declared request | MCP arguments |
| --- | --- |
| Pydantic model or object schema | The declared fields directly |
| Scalar, list, or union | `{ "input":  }` |
| No request | `{}` |

Every call returns a versioned structured result:

```json
{
  "schema_version": 1,
  "ok": true,
  "result": {
    "id": "project_123",
    "name": "Launch"
  }
}
```

Expected failures use the same envelope:

```json
{
  "schema_version": 1,
  "ok": false,
  "error": {
    "kind": "application_error",
    "code": "PROJECT_NOT_FOUND",
    "message": "Project not found"
  }
}
```

The published output schema enumerates each tool's declared application error
codes. Undeclared `AppError` values and unexpected exceptions become a generic
`failed` result. Invalid output becomes `invalid_result`; neither result
contains exception text, request input, or the invalid value.

`TOOL_MCP_PROTOCOL_VERSION` identifies this envelope and schema mapping. Tenchi
uses a new version whenever the wire shape changes.

## Run a trusted local stdio server

For a local process launched on behalf of one trusted user, the authentication
callback can return a fixed principal:

```python
from tenchi.mcp import create_tool_mcp_server

mcp = create_tool_mcp_server(
    tools=tools,
    authenticate=lambda request: local_operator,
    runner_factory=lambda user: create_user_tool_runner(user=user),
)

if __name__ == "__main__":
    mcp.run(transport="stdio")
```

The process credentials and configured principal define who the stdio client
acts as. Do not use a shared, privileged principal for untrusted clients.

## Verify the integration

Protect the transport-neutral tool contract before testing the adapter:

```shell
uv run tenchi tools --diff tools.json
uv run tenchi tools --check tools.json
```

The snapshot covers the names, JSON Schemas, declared errors, and safety
annotations this MCP server publishes. Use `--diff-ref` in pull requests so the
baseline comes from before the branch. The application MCP result envelope has
its own protocol version; Tenchi tests that wire shape separately.

Test through an in-memory MCP session so discovery, schemas, approval, and
execution use the real protocol adapter:

```python
from mcp.shared.memory import create_connected_server_and_client_session

async with create_connected_server_and_client_session(mcp) as session:
    listed = await session.list_tools()
    result = await session.call_tool(
        "projects.search",
        {"query": "launch"},
    )

assert [item.name for item in listed.tools] == ["projects.search"]
assert result.structuredContent == {
    "schema_version": 1,
    "ok": True,
    "result": [{"id": "project_123", "name": "Launch"}],
}
```

The in-memory transport still runs authentication and visibility callbacks. A
real HTTP transport additionally supplies `McpRequest.transport_request`.

---

# Testing
Source: https://tenchi.io/testing

Test behavior at the narrowest useful boundary: use cases directly for
application logic, the typed client for contract behavior, and raw HTTP for
exact envelopes and headers.

## Test use cases directly

```python
async def test_create_todo_persists() -> None:
    repository = MemoryTodoRepository()
    context = AppContext(todos=repository)

    todo = await create_todo(CreateTodo(title="Buy milk"), context)

    assert todo.completed is False
    assert await repository.get(todo.id) == todo
```

These tests need no ASGI application and make authorization and domain behavior
easy to exercise.

## Use the typed in-process client

```python
from tenchi.testing import open_client

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

assert created.body.title == "Buy milk"
assert created.headers.location.endswith(created.body.id)
```

`open_client()` runs the application lifespan and uses httpx's ASGI transport.
It provides the same contract validation as a network client.
Pass `observers=` to collect the same payload-safe `ClientOutcome` values that
a deployed client emits.

## Assert raw HTTP behavior

```python
from tenchi.testing import open_http

async with open_http(app) as http:
    response = await http.post(
        "/todos",
        json={"title": ""},
    )

assert response.status_code == 422
assert response.headers["x-tenchi-error-source"] == "framework"
assert response.json()["code"] == "VALIDATION_ERROR"
```

Use raw HTTP for malformed input, media types, request IDs, middleware,
application error envelopes, and other wire-level behavior.

## Replace ports, not framework internals

Build a fresh application around memory adapters for behavior-focused
integration tests. Use temporary real adapters when lifecycle, transactions,
or persistence are the behavior under test; the generated starter verifies
that todos survive a new application instance against the same temporary
SQLite database. Avoid monkeypatching route dispatch or internal Tenchi
functions—the purpose of these tests is to exercise the real boundary.

## Verify store adapters

Run Tenchi's conformance checks against each idempotency or rate-limit adapter
you plan to deploy:

```python
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from tenchi.idempotency import IdempotencyStore
from tenchi.testing import verify_idempotency_store

class TestClock:
    def __init__(self, value: float = 1_000.0) -> None:
        self.value = value

    def __call__(self) -> float:
        return self.value

    def advance(self, seconds: float) -> None:
        self.value += seconds

clock = TestClock()

@asynccontextmanager
async def open_store() -> AsyncGenerator[IdempotencyStore]:
    async with open_test_connection() as connection:
        try:
            yield PostgresIdempotencyStore(connection, clock=clock)
        except BaseException:
            await connection.rollback()
            raise
        else:
            await connection.commit()

async def test_postgres_idempotency_store_conforms() -> None:
    await verify_idempotency_store(open_store, advance=clock.advance)
```

Replace `open_test_connection()` and `PostgresIdempotencyStore` with your
adapter's test connection and constructor. `open_store()` must open a fresh
independently usable scope over one shared test backend. A successful scope
must persist before it exits. The adapter and the `advance(seconds)` callback
must use the same deterministic clock; the callback may be synchronous or
asynchronous.

`verify_idempotency_store()` checks the reservation, replay, conflict,
expiration, token-fencing, identity-isolation, and concurrent-winner rules.
Use `verify_rate_limit_store()` with the same factory shape for fixed-window
capacity, rejected costs, reset boundaries, policy changes, isolation, and
concurrent admission. Failures raise `StoreConformanceError` with the store
kind, case name, and reason.

These checks verify the protocol state machine. Keep separate integration tests
for behavior outside that contract: transaction rollback with application
writes, migrations, cleanup, backend clock configuration, and coordination
across the processes or hosts you deploy.

## Keep OpenAPI reproducible

Generated applications include a test that runs:

```python
from tenchi.cli import main

def test_openapi_snapshot_is_current() -> None:
    assert main(["openapi", "--check", "openapi.json"]) == 0
```

This checks exact drift. Compatibility belongs in CI against a historical
baseline; see [OpenAPI and compatibility](/openapi).

Generated applications also keep the machine-facing tool contract current:

```python
from tenchi.cli import main

def test_tool_snapshot_is_current() -> None:
    assert main(["tools", "--check", "tools.json"]) == 0
```

Run `tenchi tools --diff` against the previous snapshot before replacing it.
The exact test proves reproducibility; the historical diff proves
compatibility. See [Application tools](/tools) for the complete workflow.

---

# Production handbook
Source: https://tenchi.io/production

Tenchi keeps production infrastructure outside the framework core, but it
gives each concern a defined place. Process resources live in lifespan,
request resources live in the context scope, external systems sit behind
application-owned ports, and operational entrypoints call the same use cases
as HTTP.

This handbook turns that model into concrete recipes for a service that needs
to survive retries, concurrent writes, partial failures, and deployment
changes.

## What Tenchi owns

Tenchi provides the application seams and boundary guarantees:

| Concern | Tenchi's part | Your application's part |
| --- | --- | --- |
| Configuration | Explicit composition and lifespan state | Load and validate environment-specific values |
| Database | Lifespan and request-scoped context managers | Choose a driver, migrations, transaction isolation, and adapters |
| Authentication | Boundary hooks and public-contract metadata | Verify credentials and supply the identity port |
| Authorization | Plain use cases and pure policies | Define abilities and owner-scoped repository methods |
| Retries | Canonical fingerprints, durable store transitions, typed replay, and declared errors | Implement storage in the command transaction and choose key scope and retention |
| Rate limiting | Atomic fixed-window store protocol, policy validation, and a standard 429 | Choose authenticated scope, shared storage, transaction semantics, and edge limits |
| Inbound webhooks | Exact-body verifier bindings, required composition, request limits, and typed validation | Verify the provider protocol, attach service identity, enforce replay rules, and store event ids |
| Background work | Validated job messages, handler bindings, dispatch, and scoped contexts | Choose a queue or outbox, schedule workers, and define retry policy |
| Operational maintenance | Validated named tasks, CLI/MCP discovery, and scoped execution | Define safe dry runs, operator access, and rollout procedure |
| Observability | Finalized outcomes plus an optional OpenTelemetry bridge | Configure the SDK, transport instrumentation, exporters, alerts, and retention |
| Deployment | Health routes, request limits, deadlines, `tenchi check`, and environment preflight | Configure the ASGI server, proxy, secrets, migrations, dependency credentials, and rollout |

  Tenchi does not bundle an ORM, queue, scheduler, telemetry SDK, or settings
  package. Those libraries remain replaceable adapters. The handbook defines
  how they participate in the application lifecycle and failure model.

## Build the production boundary

Use the following order when turning a working Tenchi application into a
deployed service:

1. [Validate configuration and secrets](/configuration) once at startup.
2. [Open process resources and one transaction per request](/database).
3. Put outbound SDKs, repositories, clocks, and identity lookups behind ports.
4. Make retried commands idempotent and [defer external effects through a
   durable worker boundary](/reliability).
5. [Apply authenticated operation quotas and configure edge limits](/rate-limits).
6. [Verify inbound webhook bytes and collapse provider redeliveries](/webhooks).
7. [Emit low-cardinality request and worker telemetry](/observability).
8. [Declare read-only checks for the target environment](/preflight).
9. [Configure health, limits, middleware, and the release gate](/deployment).

Authentication and authorization have their own
[boundary-to-policy workflow](/authentication). Testing follows the same
resource model: use memory adapters for application behavior, then exercise
real transactions and lifespan in integration tests.

## Choose a consistency requirement before an adapter

A port should say what its caller needs, not which vendor implements it. A
command repository normally needs strong reads and writes in the current
transaction. A search port can explicitly permit stale results and use a read
replica. A notification port can promise only that work was durably accepted,
not that an email was delivered during the request.

Keep those meanings visible in method names and docstrings. Wiring can then
change from SQLite to PostgreSQL, an in-process adapter to a service client, or
a primary connection to a replica without silently weakening a use case.

## Define the failure owner

For every boundary, decide which component owns each outcome:

- A contract owns valid HTTP inputs, outputs, and declared application errors.
- A request context owns commit or rollback.
- A use case owns business rejection through `AppError`.
- Tenchi job primitives validate producer messages, consumer input, and
  handler results.
- A worker owns acknowledgement, retry, dead-lettering, and backoff.
- A deployment process owns migrations and compatibility with the previous
  release.
- An observer reports an outcome but never changes it.

This avoids ambiguous failures such as retrying invalid payloads forever,
committing state before its outbox record, or treating a logging outage as an
API failure.

## Production baseline

Before the first deployment, verify that the service has:

- validated startup configuration with no secret values in logs;
- graceful acquisition and cleanup for every pool and SDK client;
- commit-on-success and rollback-on-error request scopes;
- an explicit migration step and rollback-compatible deployment plan;
- idempotency for commands callers or infrastructure can retry;
- shared, atomic rate-limit storage plus gateway limits for unauthenticated
  traffic;
- exact-body verification, timestamp checks, and idempotency for inbound
  webhooks;
- retry and dead-letter rules for every background handler;
- trusted request IDs, bounded-cardinality telemetry, and actionable alerts;
- readiness checks for dependencies required to serve traffic;
- body-size limits and operation deadlines appropriate to the API;
- a read-only `tenchi preflight` gate in the target environment;
- `tenchi verify --base-ref ` for checks, strict architecture
  evidence, and OpenAPI and application-tool comparisons against one immutable
  release baseline.

The [deployment guide](/deployment) turns this list into the final release
sequence.

---

# Configuration and secrets
Source: https://tenchi.io/configuration

Load configuration once at the composition root, validate it before the
application accepts traffic, and pass the resulting values into resource
factories. Use cases and adapters should not read environment variables
directly.

## Define a validated settings model

Pydantic is already a Tenchi dependency, so a small application can validate
its environment without another package:

```python
import os
from collections.abc import Mapping
from typing import Literal

from pydantic import BaseModel, Field, SecretStr

class Settings(BaseModel):
    environment: Literal["development", "staging", "production"]
    database_url: str = Field(min_length=1)
    auth_secret: SecretStr
    request_timeout_seconds: float = Field(default=10.0, gt=0)

def load_settings(env: Mapping[str, str] = os.environ) -> Settings:
    return Settings.model_validate(
        {
            "environment": env.get("APP_ENV"),
            "database_url": env.get("APP_DATABASE_URL"),
            "auth_secret": env.get("APP_AUTH_SECRET"),
            "request_timeout_seconds": env.get(
                "APP_REQUEST_TIMEOUT_SECONDS",
                "10",
            ),
        }
    )
```

Call `load_settings()` while composing the application. A missing database URL,
missing secret, invalid environment name, or non-positive timeout then stops
startup with a Pydantic validation error.

For layered files, cloud secret managers, or custom settings sources, use a
dedicated package such as `pydantic-settings`. Keep the result as one typed
application value regardless of where the values came from.

## Wire settings into lifespan

Process-scoped resources should receive only the settings they need:

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from app.infra.database import DatabasePool, open_database_pool
from app.infra.tokens import TokenVerifier

@dataclass(frozen=True, slots=True)
class Runtime:
    database: DatabasePool
    tokens: TokenVerifier

settings = load_settings()

@asynccontextmanager
async def lifespan() -> AsyncIterator[Runtime]:
    database = await open_database_pool(settings.database_url)
    tokens = TokenVerifier(settings.auth_secret.get_secret_value())
    try:
        yield Runtime(database=database, tokens=tokens)
    finally:
        await database.close()
```

The object yielded by lifespan is passed to a one-argument context factory.
That factory creates request-scoped repositories from the pool and makes the
token verifier available to the authentication hook through the context.

Keep configuration out of `AppContext` unless a use case genuinely needs a
value as application input. Database URLs, credentials, telemetry endpoints,
and SDK options are wiring concerns; the constructed ports belong in context.

## Treat secrets as capabilities

- Store production secrets in the platform's secret manager, not in the
  repository, image, or generated OpenAPI document.
- Use `SecretStr` or an equivalent wrapper so accidental representations are
  redacted.
- Call `get_secret_value()` only where a concrete client is constructed.
- Pass a verifier or service port into context instead of passing the raw
  secret through application code.
- Rotate secrets through the deployment platform. If rotation must happen
  without restart, put the refresh behavior behind a port rather than reading
  the environment on each request.

  Request headers can also contain credentials. Log selected, redacted fields
  rather than serializing settings, `os.environ`, or complete header maps.

## Separate deploy-time and runtime configuration

Deploy-time values select infrastructure and security policy: database URLs,
issuer metadata, trusted hosts, CORS origins, telemetry exporters, and feature
rollout configuration. Runtime request data belongs in validated contract
inputs or authenticated identity.

Changing a deploy-time value should create a new process generation. This
makes rollbacks predictable and lets readiness checks verify the exact
resources that generation opened.

## Test configuration failures

Pass a plain mapping to `load_settings()` in tests:

```python
import pytest
from pydantic import ValidationError

def test_requires_the_auth_secret() -> None:
    with pytest.raises(ValidationError):
        load_settings(
            {
                "APP_ENV": "production",
                "APP_DATABASE_URL": "postgresql://db.example/app",
            }
        )
```

Test the model separately from resource-opening integration tests. A unit test
should not need the process environment or a secret manager.

---

# Databases and transactions
Source: https://tenchi.io/database

Open database pools for the process, acquire a connection for each request,
and bind every write adapter in that request to the same transaction. Tenchi's
lifespan and context scopes make those ownership boundaries explicit.

## Put one unit of work in the request context

The following SQLite shape commits only after the complete Tenchi boundary
succeeds. A hook rejection, use-case error, deadline, or response validation
failure rolls the transaction back before Tenchi creates the HTTP error:

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import aiosqlite

from app.infra.sqlite_repositories import SqliteOutbox, SqliteTodoRepository
from app.server.context import AppContext
from tenchi.server import create_app

@asynccontextmanager
async def create_context(
    database_path: str,
) -> AsyncIterator[AppContext]:
    async with aiosqlite.connect(database_path) as connection:
        await connection.execute("PRAGMA foreign_keys = ON")
        try:
            yield AppContext(
                todos=SqliteTodoRepository(connection),
                outbox=SqliteOutbox(connection),
            )
        except BaseException:
            await connection.rollback()
            raise
        else:
            await connection.commit()
```

Pass the path or a pool through lifespan:

```python
app = create_app(
    routes=routes,
    lifespan=lifespan,
    context_factory=create_context,
)
```

With a pooled driver, lifespan opens and closes the pool while
`create_context()` acquires one connection and enters the driver's transaction
context. The ownership rule stays the same.

  Repositories that participate in one command must use the request's
  connection. A process-global connection can mix concurrent requests and
  makes commit ownership ambiguous.

## Run migrations as a release step

For a service with multiple replicas, run migrations once through the
database's migration tool before new instances receive traffic. Application
startup should verify connectivity and the schema version; it should not let
every replica race to perform an uncoordinated migration.

Use an expand-and-contract rollout for changes that span application
generations:

1. Add nullable columns, new tables, or compatible indexes.
2. Deploy code that can work with both the old and expanded schema.
3. Backfill data with an observable, restartable task.
4. Switch reads and writes to the new representation.
5. Remove the old representation only after the previous application
   generation can no longer run.

A single-process SQLite deployment can perform small idempotent schema setup
in lifespan, but it still needs a lock when the API and a worker can start
together.

## State consistency in ports

Separate ports when callers need different consistency:

```python
from typing import Protocol

class TaskRepository(Protocol):
    """Writes and read-your-writes queries on the primary transaction."""

    async def get(self, task_id: str) -> Task | None: ...
    async def save(
        self,
        task: Task,
        *,
        expected_version: int,
    ) -> Task | None: ...

class TaskSearch(Protocol):
    """Staleness-tolerant listing that may use a read replica."""

    async def search(self, query: TaskQuery) -> list[Task]: ...
```

Wiring may bind `TaskSearch` to a replica, but a write use case should fetch
through `TaskRepository` when it needs read-your-writes behavior. Naming this
requirement prevents a later infrastructure change from silently weakening
the application.

## Prevent lost updates

Use optimistic concurrency when two callers can update the same resource:

1. Return a strong `ETag` derived from the stored version.
2. Require that value in `If-Match` on writes.
3. Update with `WHERE id = ? AND version = ?` and increment the version in the
   same statement.
4. Return a declared `428` when the precondition is missing and `412` when it
   is stale.

The repository must make the final comparison atomically:

```python
cursor = await connection.execute(
    "UPDATE tasks "
    "SET title = ?, version = version + 1 "
    "WHERE id = ? AND version = ? "
    "RETURNING id, title, version",
    (task.title, task.id, expected_version),
)
row = await cursor.fetchone()
return row_to_task(row) if row is not None else None
```

A read followed by an unconditional update is not sufficient: another writer
can commit between those statements.

## Test the transaction boundary

Use direct use-case tests with memory adapters for application behavior. Add
integration tests with the real database adapter for:

- commit after a successful response;
- rollback after `AppError`, unexpected failure, cancellation, or response
  validation failure;
- unique constraints and foreign keys;
- concurrent idempotency claims and optimistic updates;
- migration from every schema version you still deploy;
- read-replica behavior when a port permits stale data.

Run HTTP integration tests with [`open_client()` or `open_http()`](/testing) so
lifespan and request scopes execute exactly as they do under an ASGI server.

---

# Retry-safe operations
Source: https://tenchi.io/idempotency

Use Tenchi's idempotency primitive when the same logical command can arrive
more than once: a client retries after a timeout, a webhook is redelivered, or
a worker restarts work whose outcome it did not observe.

`run_idempotently()` gives one reservation permission to execute and returns
the stored typed result to matching retries. Your application supplies the
durable store and decides which transaction contains the reservation,
application writes, and completed result.

## Declare the HTTP boundary

For an HTTP command, validate the key as an ordinary request header and
declare both standard idempotency errors:

```python
from pydantic import BaseModel, Field

from tenchi.contracts import contract
from tenchi.idempotency import (
    IDEMPOTENCY_CONFLICT,
    IDEMPOTENCY_IN_PROGRESS,
)

class CreateTaskHeaders(BaseModel):
    idempotency_key: str = Field(
        alias="Idempotency-Key",
        min_length=1,
        max_length=128,
        pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$",
    )

create_task_contract = contract(
    method="POST",
    path="/tasks",
    headers=CreateTaskHeaders,
    request=CreateTask,
    response=Task,
    status=201,
    errors=(IDEMPOTENCY_CONFLICT, IDEMPOTENCY_IN_PROGRESS),
)
```

Missing or malformed headers fail normal request validation. A reused key with
different input raises `IDEMPOTENCY_CONFLICT`. Matching work that still owns
an active reservation raises `IDEMPOTENCY_IN_PROGRESS`, optionally with
`Retry-After`.

These are application errors under Tenchi's
[honesty rule](/errors#the-honesty-rule).
If the contract does not declare them, they become framework-owned 500
responses instead of undocumented 409 responses.

## Put the store in the application context

Add `IdempotencyStore` beside the other ports required by your use cases:

```python
from dataclasses import dataclass

from tenchi.idempotency import IdempotencyStore

@dataclass(frozen=True, slots=True)
class AppContext:
    tasks: TaskRepository
    idempotency: IdempotencyStore
    user: User | None = None
```

The protocol has three async operations:

| Method | Required behavior |
| --- | --- |
| `reserve()` | Atomically return one reservation, a completed replay, a fingerprint conflict, or an active matching reservation |
| `complete()` | Store the serialized result only when the active, unexpired reservation token still owns the key; fail if it cannot be stored |
| `abandon()` | Release only the matching active token; repeated calls must be safe |

An adapter uses `(namespace, scope, key)` as its unique identity. It stores the
fingerprint separately, fences completion with the opaque reservation token,
and expires abandoned reservations after `reservation_ttl`.

Use `MemoryIdempotencyStore` for use-case and in-process HTTP tests. It is
concurrency-safe within one process and accepts a deterministic clock:

```python
from tenchi.idempotency import MemoryIdempotencyStore

now = 1_000.0
store = MemoryIdempotencyStore(clock=lambda: now)
```

Share one instance across every context created by the test application. The
memory store loses all records on restart and cannot coordinate multiple
processes or hosts, so it is not a production durability boundary.

`reservation_ttl` is a lease, not an operation deadline. Set it longer than
the longest permitted operation, including expected dependency latency. The
primitive does not renew reservations automatically; if a lease expires while
work is still running, a later caller may reserve and execute the operation.

  When an operation writes to a database, persist its reservation, application
  writes, and completed replay in that database transaction. A separate cache
  can publish a replay before the database commits or lose the reservation
  after the command commits. Tenchi cannot make two independent systems
  atomic.

  A database transaction cannot include an email provider, webhook receiver,
  or unrelated remote API. If that system accepts a call and the local
  operation then fails, abandoning the reservation permits a retry to call it
  again. Pass the same idempotency key to a downstream service that supports
  one, or commit the effect through a [transactional outbox](/reliability#commit-deferred-effects-with-state).

## Run the use case once

Build the fingerprint from validated input, then wrap only the state-changing
operation:

```python
from tenchi.idempotency import fingerprint, run_idempotently

async def create_task(
    headers: CreateTaskHeaders,
    request: CreateTask,
    context: AppContext,
) -> Task:
    user = require_user(context.user)
    project = await context.projects.get(request.project_id)
    ensure_can_write_project(user, project, project_id=request.project_id)

    async def create() -> Task:
        return await context.tasks.create(
            project_id=request.project_id,
            title=request.title,
        )

    return await run_idempotently(
        context.idempotency,
        namespace="tasks.create",
        scope=user.id,
        key=headers.idempotency_key,
        fingerprint=fingerprint(request, annotation=CreateTask),
        result_type=Task,
        operation=create,
        completed_ttl=24 * 60 * 60,
    )
```

The namespace is a stable dotted `snake_case` operation name. The scope must
prevent one actor or tenant from replaying another's result. Use a global scope
only when every caller should share the same key namespace.

`fingerprint()` validates against `annotation=` before hashing canonical JSON.
Equivalent validated values therefore share a fingerprint even when mapping
order or coercible input spelling differs.

`set` and `frozenset` values are rejected because their iteration order is not
a stable serialization boundary. Convert them to a sorted tuple or list before
fingerprinting.

Include every validated value that can change the operation's result: path,
query, headers other than the idempotency key, and body. When a command has
more than one input model, create a small frozen dataclass or Pydantic model
that contains those values and fingerprint that combined input.

## Understand each outcome

| Store decision | `run_idempotently()` behavior |
| --- | --- |
| Reservation | Calls the operation, validates its result, stores serialized JSON, and returns the validated value |
| Replay | Validates the stored JSON against `result_type` and returns it without calling the operation |
| Conflict | Raises `AppError(IDEMPOTENCY_CONFLICT)` |
| In progress | Raises `AppError(IDEMPOTENCY_IN_PROGRESS)` and preserves the store's optional `Retry-After` value |

If the operation fails, is cancelled, returns an invalid result, or cannot
complete its record, Tenchi attempts to abandon the reservation before
propagating the original failure. Reservation expiration remains the recovery
path if a process stops before cleanup runs.

Store enough result data to reproduce the complete caller-visible success.
For an HTTP route with status-dependent presentation or response headers, the
replayed use-case result must let the presenter derive the same status, body,
and headers.

## Choose retention and compatibility

`completed_ttl=None` retains successful replays until the adapter or an
operational cleanup removes them. A positive TTL starts when the result is
completed. Keep records at least as long as callers may legitimately retry a
key.

A replay is validated against the current `result_type`. Deployments must
remain able to read stored results for the chosen retention window. When a
result schema changes incompatibly, migrate stored values, shorten and drain
the old retention window before deployment, or version the operation
namespace.

Do not store request bodies, credentials, or provider exceptions in an
idempotency record. The fingerprint is sufficient to compare input; the
completed record needs only the validated result required for replay.
SHA-256 is not anonymization: a fingerprint of low-entropy secret data can
still be guessed. Exclude secrets from the logical input whenever they do not
affect the operation.

## Retry from a client

Generate one key for one logical command and reuse it across that command's
transport retries:

```python
from uuid import uuid4

from tenchi.retries import retry_policy

key = uuid4().hex
headers = CreateTaskHeaders(idempotency_key=key)

task = await client.call(
    create_task_contract,
    headers=headers,
    request=request,
    retry=retry_policy(
        max_attempts=3,
        total_timeout_seconds=5,
        allow_unsafe_methods=True,
    ),
)
```

`allow_unsafe_methods=True` records the deliberate decision to repeat POST. The
idempotency key makes those attempts one logical command. Do not generate the
key inside retry machinery or per attempt; a new key describes new work and
permits the operation to run again. See the [typed client](/client#retry-a-logical-call)
for declared-error retries, `Retry-After`, deadlines, and attempt outcomes.

The taskboard example contains a transactional SQLite store and uses the
built-in memory store in tests. The SQLite store shares the request-scoped
connection with task creation, so concurrent requests either create once or
replay the committed result. Run
[`verify_idempotency_store()`](/testing#verify-store-adapters) against your
production adapter before relying on it, then separately test rollback with
the application writes it protects.

---

# Rate limiting
Source: https://tenchi.io/rate-limits

Use Tenchi's rate-limit primitive when an authenticated actor or tenant may
perform an application operation only a fixed number of times in a window.
The use case chooses the policy and scope; a replaceable store performs one
atomic consume.

An exhausted window raises the standard `RATE_LIMITED` application error with
HTTP status `429` and a `Retry-After` header.

## Declare the HTTP outcome

Declare `RATE_LIMITED` on every contract that may expose it:

```python
from tenchi.contracts import contract
from tenchi.rate_limits import RATE_LIMITED

create_task_contract = contract(
    method="POST",
    path="/tasks",
    request=CreateTask,
    response=Task,
    status=201,
    errors=(RATE_LIMITED,),
)
```

Tenchi documents the response and `Retry-After` header in OpenAPI. If the
contract does not declare the error, the
[honesty rule](/errors#the-honesty-rule) turns it into a framework-owned 500
instead of exposing an undocumented 429.

## Put shared storage in the context

Add the store protocol to the application context:

```python
from dataclasses import dataclass

from tenchi.rate_limits import RateLimitStore

@dataclass(frozen=True, slots=True)
class AppContext:
    tasks: TaskRepository
    rate_limits: RateLimitStore
    user: User | None = None
```

The store has one async method:

```python
from typing import Protocol

from tenchi.rate_limits import RateLimitDecision

class RateLimitStore(Protocol):
    async def consume(
        self,
        *,
        namespace: str,
        scope: str,
        limit: int,
        window_seconds: float,
        cost: int,
    ) -> RateLimitDecision: ...
```

`consume()` must be atomic. The first accepted cost opens a fixed window for
the `(namespace, scope)` identity. Later costs are accepted while their sum is
at most `limit`. A rejection does not consume capacity. At the reset boundary,
the next accepted cost opens a new window. Changing the limit or window starts
a new window immediately.

The store returns either:

- `RateLimitPermit(limit, remaining, reset_after_seconds)` for accepted work;
- `RateLimitExceeded(limit, retry_after_seconds)` for rejected work.

`enforce_rate_limit()` validates those values before trusting them. A malformed
adapter response raises `RateLimitStoreError` rather than producing a misleading
allowance or `Retry-After` value.

## Scope from verified identity

Enforce the policy after authentication, using identity derived from the
application context:

```python
from tenchi.rate_limits import enforce_rate_limit

async def create_task(request: CreateTask, context: AppContext) -> Task:
    user = require_user(context.user)
    await enforce_rate_limit(
        context.rate_limits,
        namespace="tasks.create",
        scope=user.id,
        limit=5,
        window_seconds=60,
    )
    return await context.tasks.create(
        project_id=request.project_id,
        title=request.title,
    )
```

The namespace is a stable dotted `snake_case` operation name. The scope is the
actor, tenant, credential, or other application-owned subject that shares the
allowance.

Do not use an account, tenant, or owner ID copied directly from request input.
An attacker could select another subject's scope and exhaust its capacity.
Authenticate first and derive the scope from verified context.

Use separate namespaces when operations have separate allowances. Use a tenant
scope when every user in that tenant should share one budget. Prefixing values
such as `user:alice` and `tenant:acme` can keep mixed scope kinds explicit.

## Decide what one cost represents

Place the consume according to what the policy limits.

For a transport-attempt limit, consume before the application operation in
storage that commits independently. Invalid or failed attempts still cost
capacity.

For a logical-operation quota, combine it with idempotency so transport retries
do not consume the allowance repeatedly:

```python
async def create_task(
    headers: CreateTaskHeaders,
    request: CreateTask,
    context: AppContext,
) -> Task:
    user = require_user(context.user)

    async def create() -> Task:
        await enforce_rate_limit(
            context.rate_limits,
            namespace="tasks.create",
            scope=user.id,
            limit=5,
            window_seconds=60,
        )
        return await context.tasks.create(
            project_id=request.project_id,
            title=request.title,
        )

    return await run_idempotently(
        context.idempotency,
        namespace="tasks.create",
        scope=user.id,
        key=headers.idempotency_key,
        fingerprint=fingerprint(request, annotation=CreateTask),
        result_type=Task,
        operation=create,
    )
```

Here, a completed idempotent replay bypasses `create()` and costs nothing.
Taskboard stores the rate-limit window in the same transaction as task creation,
so a rollback also restores the capacity.

Choose independent storage instead when failed operation attempts must count.
Write that choice down with the policy; transaction placement changes observable
behavior during failures and retries.

## Apply weighted costs

Use `cost=` when operations consume different amounts from one shared budget:

```python
await enforce_rate_limit(
    context.rate_limits,
    namespace="reports.export",
    scope=f"tenant:{tenant.id}",
    limit=100,
    window_seconds=60 * 60,
    cost=10,
)
```

`limit` and `cost` must be positive integers, and cost cannot exceed the limit.
`window_seconds` must be finite and greater than zero. Tenchi validates the
policy before calling the store.

## Test with the memory store

`MemoryRateLimitStore` is concurrency-safe within one process and accepts a
clock for deterministic boundary tests:

```python
import pytest

from tenchi.errors import AppError
from tenchi.rate_limits import (
    RATE_LIMITED,
    MemoryRateLimitStore,
    enforce_rate_limit,
)

now = 1_000.0
store = MemoryRateLimitStore(clock=lambda: now)

await enforce_rate_limit(
    store,
    namespace="tasks.create",
    scope="alice",
    limit=1,
    window_seconds=60,
)

with pytest.raises(AppError) as excinfo:
    await enforce_rate_limit(
        store,
        namespace="tasks.create",
        scope="alice",
        limit=1,
        window_seconds=60,
    )

assert excinfo.value.definition == RATE_LIMITED
assert excinfo.value.headers == {"Retry-After": "60"}

now += 60
permit = await enforce_rate_limit(
    store,
    namespace="tasks.create",
    scope="alice",
    limit=1,
    window_seconds=60,
)
assert permit.remaining == 0
```

Share one memory store across every context created by the test application.
Creating a new store per request creates a new empty allowance and does not
test rate limiting.

  `MemoryRateLimitStore` cannot coordinate multiple processes or hosts and
  loses all windows on restart. Use it for use-case tests, in-process HTTP
  tests, and local development only.

## Implement production storage

A production adapter must atomically read and update one
`(namespace, scope)` window. Use a conditional database write, a Redis script,
or another operation that cannot admit two concurrent callers beyond the
limit.

Run [`verify_rate_limit_store()`](/testing#verify-store-adapters) against the
adapter with a shared test backend and deterministic clock. It exercises
capacity, rejected-cost behavior, resets, policy changes, identity isolation,
and concurrent admission. Keep separate integration tests for transaction
rollback and coordination across the processes or hosts you deploy.

Store at least:

- namespace and scope;
- configured limit and window duration;
- consumed cost;
- reset instant.

Delete expired windows during consumption or through operational cleanup.
Use the storage system's clock when several hosts share the store, so host clock
skew cannot lengthen or shorten a caller's window.

Fixed windows can admit traffic near both sides of a reset boundary. Use an API
gateway with a rolling-window, token-bucket, or other algorithm when that burst
shape is unacceptable.

## Keep edge protection at the edge

Tenchi's primitive protects authenticated application operations. It is not a
denial-of-service boundary: the application has already accepted the
connection, matched a route, and usually authenticated the caller.

Configure your reverse proxy, load balancer, CDN, or API gateway for:

- unauthenticated and per-IP request floods;
- connection and request-rate ceilings;
- distributed bot and abuse controls;
- global protection before application workers are occupied.

Keep application limits for rules that need verified users, tenants, plans,
operation names, or domain-specific weighted costs. Many deployed services need
both layers.

---

# Signed webhooks
Source: https://tenchi.io/webhooks

Signed webhooks authenticate an external service against the exact bytes it
sent. Tenchi verifies those bytes before parsing the request, then applies the
contract's normal Pydantic validation and invokes the use case with an
enriched application context.

## Declare the delivery contract

Model the provider payload as ordinary application input. Include the
provider's stable delivery or event identifier when it offers one:

```python
from pydantic import BaseModel, Field

class MemberAddedWebhook(BaseModel):
    event_id: str = Field(min_length=1, max_length=200)
    project_id: str
    project_name: str
    user_id: str
```

Declare the expected verification failure and mark the contract with
`webhook=True`:

```python
from tenchi.contracts import contract
from tenchi.errors import ErrorDef
from tenchi.idempotency import IDEMPOTENCY_CONFLICT, IDEMPOTENCY_IN_PROGRESS

invalid_webhook = ErrorDef(
    code="INVALID_WEBHOOK",
    status=401,
    message="Webhook verification failed",
)
unauthorized = ErrorDef(
    code="UNAUTHORIZED",
    status=401,
    message="Unauthorized",
)

member_added_webhook_contract = contract(
    method="POST",
    path="/webhooks/member-added",
    request=MemberAddedWebhook,
    status=204,
    errors=(
        invalid_webhook,
        unauthorized,
        IDEMPOTENCY_CONFLICT,
        IDEMPOTENCY_IN_PROGRESS,
    ),
    name="webhooks.member_added",
    public=True,
    webhook=True,
    max_request_bytes=64 * 1024,
    timeout=10,
)
```

`webhook=True` is a composition requirement, not documentation alone.
`create_app()` refuses to build the application unless the contract has a
verifier binding. OpenAPI marks the operation with
`x-tenchi-webhook: true`.

Set `public=True` when a general authentication hook should exempt the route.
The webhook verifier still authenticates it. If the endpoint also requires
your application's ordinary authentication, leave `public=False`.

## Keep service authorization in the use case

Add optional service identity to the application context:

```python
from dataclasses import dataclass

from tenchi.idempotency import IdempotencyStore

@dataclass(frozen=True, slots=True)
class AppContext:
    notifications: NotificationLog
    idempotency: IdempotencyStore
    service: str | None = None
```

The verifier attaches identity at the HTTP boundary. The use case still
asserts it, so direct calls and tests cannot bypass the rule:

```python
from tenchi.errors import AppError
from tenchi.idempotency import fingerprint, run_idempotently

def require_service(service: str | None, expected: str) -> str:
    if service != expected:
        raise AppError(unauthorized)
    return expected

async def receive_member_added_webhook(
    request: MemberAddedWebhook,
    context: AppContext,
) -> None:
    service = require_service(context.service, "member-directory")

    async def record_notification() -> None:
        await context.notifications.record(
            user_id=request.user_id,
            message=f"You were added to project {request.project_name!r}",
        )

    await run_idempotently(
        context.idempotency,
        namespace="webhooks.member_added",
        scope=service,
        key=request.event_id,
        fingerprint=fingerprint(request, annotation=MemberAddedWebhook),
        result_type=type(None),
        operation=record_notification,
        completed_ttl=7 * 24 * 60 * 60,
    )
```

Declare every `AppError` this use case or its verifier may raise. Undeclared
errors follow Tenchi's normal honesty rule and become framework-owned 500
responses.

## Verify the exact body

Keep secrets and provider SDKs at the server composition boundary. This
example verifies a `sha256=` HMAC header:

```python
import hashlib
import hmac
from dataclasses import replace

from tenchi.errors import AppError
from tenchi.webhooks import WebhookRequest, webhook

def create_member_added_webhook(secret: bytes):
    if not secret:
        raise ValueError("webhook secret must not be empty")

    def verify(
        request: WebhookRequest,
        context: AppContext,
    ) -> AppContext:
        signatures = request.header_values.get("x-webhook-signature", ())
        if len(signatures) != 1:
            raise AppError(invalid_webhook)
        presented = signatures[0]
        expected = "sha256=" + hmac.new(
            secret,
            request.body,
            hashlib.sha256,
        ).hexdigest()

        try:
            valid = hmac.compare_digest(
                presented.encode("ascii"),
                expected.encode("ascii"),
            )
        except UnicodeEncodeError:
            valid = False
        if not valid:
            raise AppError(invalid_webhook)

        return replace(context, service="member-directory")

    return webhook(member_added_webhook_contract, verify)
```

`WebhookRequest.body` is immutable and preserves JSON whitespace, field
ordering, and the final newline. `headers` is a read-only mapping with
lowercased names and the last value for ordinary access. `header_values`
preserves every repeated value in arrival order so a verifier can reject
ambiguous signature headers. The request also includes the matched contract
and Tenchi request ID.

Use the provider's official verifier when its signature format includes
multiple signatures, custom timestamp encoding, certificate validation, or
key rotation. Pass `request.body` and the required raw headers directly to
that verifier; do not serialize a parsed model back into JSON.

## Bind the verifier

Load the secret from validated process configuration and bind the verifier
when composing the application:

```python
app = create_app(
    routes=routes,
    context_factory=create_context,
    hooks=(authenticate,),
    webhooks=(
        create_member_added_webhook(settings.member_webhook_secret),
    ),
)
```

Ordinary hooks run first. Tenchi then checks the declared media type and body
size, reads the body once, runs the matching webhook verifier, validates the
path, query, header, and request models, and invokes the use case. The route
timeout includes verifier work and cancellation cleanup.

Verifiers may be synchronous or asynchronous. Return `None` to keep the
current context, or return an enriched context. An expected rejection should
raise `AppError`; an unexpected verifier failure is logged and returned as a
framework-owned 500 without exposing the exception.

## Prevent replayed effects

A valid signature proves authenticity and integrity. By itself, it does not
prove that the delivery is new.

- Enforce the provider's signed timestamp tolerance when its protocol
  supplies one.
- Store the provider event ID through
  [`run_idempotently()`](/idempotency), scoped to the provider or tenant.
- Keep the idempotency transition in the same transaction as application
  writes.
- Retain completed event IDs for at least the provider's documented retry
  window.

If processing must continue after the HTTP acknowledgement, commit a
transactional outbox record and let a worker own retry and dead-letter
behavior. Do not start untracked background work from the request.

  Webhook bodies and headers can contain personal data, credentials, or
  provider signatures. Log the contract name, request ID, bounded outcome,
  and application-owned event identifier only after applying your redaction
  policy.

## Test the wire bytes

Compute the signature over the same bytes sent by the test client:

```python
body = (
    b'{"event_id":"evt_1","project_id":"p1",'
    b'"project_name":"Launch","user_id":"u1"}'
)
signature = "sha256=" + hmac.new(
    secret,
    body,
    hashlib.sha256,
).hexdigest()

response = await http.post(
    "/webhooks/member-added",
    content=body,
    headers={
        "content-type": "application/json",
        "x-webhook-signature": signature,
    },
)
assert response.status_code == 204
```

Also test a bad signature, an invalid payload with a bad signature, a repeated
event ID, and a body above the contract's size ceiling. A bad signature wins
over Pydantic validation; oversized and unsupported-media requests are
rejected before application verification runs.

---

# Retries and background work
Source: https://tenchi.io/reliability

Networks retry. Clients time out after the server commits. Workers crash after
an external service accepts a call. Design commands and deferred work around
those facts instead of treating retries as exceptional.

## Make commands idempotent

Use [`run_idempotently()`](/idempotency) around commands that callers or
infrastructure can retry. Scope the caller-provided key to an actor or tenant,
then store three values in the same transaction as the command:

- the scoped key;
- a stable fingerprint of the validated input;
- enough of the original success to reproduce its body, status, and headers.

Tenchi applies these outcomes through the application-supplied
`IdempotencyStore`:

| Existing record | Result |
| --- | --- |
| No record | Claim the key, perform the command, store the response, and commit together |
| Same key and fingerprint | Return the original successful response |
| Same key and different fingerprint | Raise the declared `IDEMPOTENCY_CONFLICT` error |
| Same key and matching work still active | Raise the declared `IDEMPOTENCY_IN_PROGRESS` error |

Build the fingerprint from validated data:

```python
from app.features.tasks.schemas import CreateTask
from tenchi.idempotency import fingerprint

request_fingerprint = fingerprint(request, annotation=CreateTask)
```

The repository must claim the key with a unique constraint and create the
resource in one transaction. A process-local lock or cache does not protect
multiple workers and loses history on restart.

Decide and document a retention period. Do not delete an idempotency record
while callers can still legitimately retry its key.

See [Make operations safe to retry](/idempotency) for the store protocol,
typed replay, expiration, cancellation, contract errors, and client retry
workflow.

## Commit deferred effects with state

Do not send email, publish a message, or call a webhook between a database
write and commit. If the process fails in that gap, application state and the
external effect disagree.

Declare the message and keep the outbox port transport-shaped:

```python
from typing import Protocol

from tenchi.jobs import job

member_added_job = job(
    "projects.member_added",
    request=MemberAdded,
    result=None,
)

class Outbox(Protocol):
    async def enqueue(self, *, job: str, payload_json: bytes) -> None: ...
```

The concrete outbox adapter writes through the same transaction as the domain
repositories. The use case changes state and enqueues the work before
returning:

```python
from tenchi.jobs import job_message

saved = await context.projects.save(updated)
message = job_message(
    member_added_job,
    MemberAdded(
        project_id=saved.id,
        project_name=saved.name,
        user_id=request.user_id,
    ),
)
await context.outbox.enqueue(
    job=message.name,
    payload_json=message.payload_json,
)
return saved
```

`job_message()` validates the producer value and serializes compact JSON
before the transaction stores it. The payload carries the facts needed for
delivery; a worker should not have to re-read mutable state merely to
reconstruct what happened.

  The outbox makes the state change and work record atomic. Delivery is
  normally at least once: a worker can crash after the external system accepts
  the call but before the outbox acknowledgement commits. Use an idempotency
  key at the downstream service or make the consumer idempotent.

## Validate work at the worker boundary

Bind job declarations to plain async use cases at the composition root:

```python
from app.features.projects.jobs import member_added_job
from app.features.projects.use_cases.notify_member_added import (
    notify_member_added,
)
from tenchi.jobs import create_job_dispatcher, job_group, job_handler

jobs = job_group(job_handler(member_added_job, notify_member_added))
dispatcher = create_job_dispatcher(jobs=jobs)

await dispatcher.dispatch(
    entry.job,
    payload_json=entry.payload_json,
    context=context,
)
```

`job_handler()` checks request and return annotations at composition.
`dispatch()` rejects unknown names, validates stored JSON, invokes the handler,
and validates its result. The worker still owns acknowledgement and retry
behavior; the dispatcher deliberately does not choose a queue, backoff policy,
or dead-letter store. See [Background jobs](/jobs) for the complete producer,
composition, dispatch, and observer API.

## Classify failures before retrying

Use three terminal paths for each delivery attempt:

| Outcome | Transaction and queue action |
| --- | --- |
| Delivered | Commit application writes and acknowledgement together |
| Deterministic failure | Roll back partial writes, then dead-letter with a bounded error |
| Transient infrastructure failure | Roll back the claim, apply bounded backoff, and retry |

Unknown job names, `ValidationError`, and `JobResultError` are deterministic
for the same record. Classify each `AppError` code by its semantics: a rejected
business rule may be permanent, while a declared rate-limit error may be
retryable after its advertised delay. Retrying permanent failures forever
blocks useful work behind a poison message. Connection failures and dependency
timeouts may also be transient.

Set a maximum attempt count or age for transient failures, add jitter to
backoff, and alert on queue age, repeated retries, and dead-letter growth.
Preserve enough metadata to replay a corrected handler safely.

## Instrument every non-HTTP entrypoint

Pass `use_case_observers=` to `create_job_dispatcher()` to report the operation,
duration, result category, and stable application error code. Job outcomes use
the `"job"` entrypoint. Workers still own attempt counts, queue latency, and
correlation metadata because those values belong to the transport. Carry trace
or request correlation in the message envelope, not inside business payloads
that should remain stable.

See [Workers and scripts](/execution) for the `execute()` failure taxonomy and
context rules, and [Observability and audit](/observability) for telemetry
conventions.

---

# Observability and audit
Source: https://tenchi.io/observability

Observe every boundary with stable operation names, trusted correlation IDs,
bounded-cardinality fields, and explicit redaction. Keep operational telemetry
separate from durable audit records.

## Export outcomes with OpenTelemetry

Install Tenchi's optional API integration and an OpenTelemetry SDK:

```sh
uv add "tenchi[otel]" opentelemetry-sdk
```

Configure the SDK, resource, readers, span processors, and exporter for your
deployment before creating the observers. Tenchi deliberately does not choose
an exporter or set process-global providers. The
[OpenTelemetry Python exporter guide](https://opentelemetry.io/docs/languages/python/exporters/)
shows console and OTLP configurations.

```python
from tenchi.client import Client
from tenchi.opentelemetry import create_opentelemetry_observers
from tenchi.server import create_app

telemetry = create_opentelemetry_observers()

app = create_app(
    routes=routes,
    context_factory=create_context,
    observers=(telemetry.request,),
    use_case_observers=(telemetry.use_case,),
)

async with Client(
    base_url="https://inventory.example.com",
    observers=(telemetry.client,),
    attempt_observers=(telemetry.client_attempt,),
) as client:
    inventory = await client.call(get_inventory_contract)
```

The factory uses the process-global providers by default. Pass
`tracer_provider=` or `meter_provider=` when the application keeps providers
explicit instead:

```python
telemetry = create_opentelemetry_observers(
    tracer_provider=tracer_provider,
    meter_provider=meter_provider,
)
```

Use the same `telemetry.use_case` observer with every non-HTTP entrypoint:

```python
from tenchi.execution import execute
from tenchi.jobs import create_job_dispatcher
from tenchi.tasks import create_task_runner

dispatcher = create_job_dispatcher(
    jobs=jobs,
    use_case_observers=(telemetry.use_case,),
)

runner = create_task_runner(
    tasks=tasks,
    context_factory=create_context,
    use_case_observers=(telemetry.use_case,),
)

await execute(
    rebuild_index,
    context=create_context,
    use_case_observers=(telemetry.use_case,),
)
```

The integration emits the following instruments. Each duration histogram also
provides an event count through its histogram count; the explicit counters make
throughput queries independent of histogram configuration.

| Instrument | Kind | Unit |
| --- | --- | --- |
| `tenchi.http.server.requests` | Counter | `{request}` |
| `tenchi.http.server.request.duration` | Histogram | `s` |
| `tenchi.use_case.invocations` | Counter | `{invocation}` |
| `tenchi.use_case.duration` | Histogram | `s` |
| `tenchi.http.client.calls` | Counter | `{call}` |
| `tenchi.http.client.call.duration` | Histogram | `s` |
| `tenchi.http.client.attempts` | Counter | `{attempt}` |
| `tenchi.http.client.attempt.duration` | Histogram | `s` |

Metric dimensions are limited to static contract or use-case names, declared
path templates, HTTP methods and statuses, bounded outcome classifications,
entrypoint names, error ownership, and the retry decision. Application error
codes, attempt numbers, maximum attempts, and selected retry delays are useful
on individual spans but are intentionally excluded from metrics.

The observer spans are completed `INTERNAL` logical-operation spans. They
inherit the current trace context and use each outcome's UTC `completed_at`
timestamp plus Tenchi's measured duration, but they are not active while
application or transport work runs. Use standard ASGI, httpx, database, and
queue instrumentation for active transport spans and automatic context
propagation. The active transport spans and Tenchi's finalized logical spans
then describe complementary parts of the same trace.

The integration depends only on `opentelemetry-api`. Without an
application-configured SDK, its providers are no-ops. The application owns
provider flushing and shutdown in its process lifespan.

## Write a custom outcome observer

Register a synchronous or asynchronous observer with `create_app()`:

```python
import logging

from tenchi.client import Client, ClientOutcome
from tenchi.execution import UseCaseOutcome
from tenchi.server import RequestOutcome, create_app

logger = logging.getLogger("app.requests")

def observe_request(outcome: RequestOutcome) -> None:
    logger.info(
        "request.complete",
        extra={
            "operation": outcome.request.contract.name,
            "method": outcome.request.method,
            "status_code": outcome.status_code,
            "duration_seconds": outcome.duration_seconds,
            "completed_at": outcome.completed_at.isoformat(),
            "error_source": outcome.error_source,
            "request_id": outcome.request.request_id,
        },
    )

def observe_use_case(outcome: UseCaseOutcome) -> None:
    operation = str(
        getattr(outcome.use_case, "__name__", type(outcome.use_case).__name__)
    )
    logger.info(
        "use_case.complete",
        extra={
            "operation": operation,
            "entrypoint": outcome.entrypoint,
            "status": outcome.status,
            "duration_seconds": outcome.duration_seconds,
            "completed_at": outcome.completed_at.isoformat(),
            "error_code": outcome.error_code,
        },
    )

def observe_client(outcome: ClientOutcome) -> None:
    logger.info(
        "client.complete",
        extra={
            "operation": outcome.contract.name,
            "status": outcome.status,
            "status_code": outcome.status_code,
            "duration_seconds": outcome.duration_seconds,
            "completed_at": outcome.completed_at.isoformat(),
            "error_code": outcome.error_code,
        },
    )

app = create_app(
    routes=routes,
    context_factory=create_context,
    observers=[observe_request],
    use_case_observers=[observe_use_case],
)
```

The outcome arrives after the request context has closed and the response is
finalized. It therefore describes the status the caller receives and includes
time spent committing or rolling back the request scope. `completed_at` records
when that measured boundary finished even if an earlier observer delays
delivery.

Observer failures are logged and isolated from the response and later
observers. Keep observers fast: they still run before the ASGI endpoint
returns. Export through a non-blocking logging or telemetry pipeline.

## Observe use cases across entrypoints

`UseCaseOutcome` reports every use case that was actually invoked through HTTP,
a background job, an operational task, an application tool, or `execute()`:

| Field | Meaning |
| --- | --- |
| `use_case` | The invoked async function |
| `entrypoint` | `http`, `execute`, `job`, `task`, or `tool` |
| `status` | `succeeded`, `app_error`, `failed`, or `cancelled` |
| `duration_seconds` | Time spent inside the use-case function |
| `error_code` | The stable `AppError` code for `app_error`; otherwise `None` |
| `completed_at` | UTC completion time captured when the function returned or raised |

Pass the observer to `create_app(use_case_observers=...)` for HTTP,
`execute(use_case_observers=...)` for workers, schedules, scripts, and direct
calls, or `create_task_runner(use_case_observers=...)` for operational tasks.
Background dispatchers accept the same observer through
`create_job_dispatcher(use_case_observers=...)`. Application tool runners use
`create_tool_runner(use_case_observers=...)`.
It runs after the surrounding context scope closes, so transactional cleanup
finishes before telemetry is exported. Input values, return values, and
exception objects are never included.

Keep use-case observers fast too. HTTP, tasks, tools, and `execute()` all wait
for the observer chain after scope cleanup.

Validation, hook, and context-acquisition failures that happen before
invocation do not produce a use-case outcome. Conversely, a use case that
returns successfully still reports `succeeded` if later response validation or
transaction cleanup makes the HTTP request fail. Compare it with
`RequestOutcome` to distinguish application behavior from the complete HTTP
boundary.

`app_error` means the use case raised `AppError`; it does not bypass Tenchi's
error honesty rule. An undeclared application error still becomes a
framework-owned HTTP 500 in `RequestOutcome`.

When an HTTP deadline cancels the function, the use-case status is `cancelled`
and the request status is 504. If the function catches that cancellation and
returns, the use-case status is `succeeded`, but the expired HTTP deadline
still owns the response and returns 504.

## Observe outbound contract calls

`ClientOutcome` reports one finalized outcome for every typed client call:

```python
async with Client(
    base_url="https://service.example.com",
    observers=[observe_client],
) as client:
    inventory = await client.call(get_inventory_contract)
```

| Field | Meaning |
| --- | --- |
| `contract` | The contract used by the call |
| `status` | `succeeded`, `app_error`, `unexpected_response`, `transport_error`, `timed_out`, `failed`, or `cancelled` |
| `status_code` | The HTTP status when a response arrived; otherwise `None` |
| `duration_seconds` | Local preparation, transport, retry backoff, and response-validation time; observer work is excluded |
| `error_code` | The declared remote `AppError` code for `app_error`; otherwise `None` |
| `attempts` | Number of transport attempts made for the logical call |
| `completed_at` | UTC completion time captured before logical-call observers run |

The outcome is emitted for local validation failures before any request leaves
the process as well as for completed transport attempts. A response whose
status or wire data violates the contract reports `unexpected_response`.
`failed` covers failures without a response that are neither an httpx
transport error nor cancellation, such as invalid local input or contract
configuration.

Client outcomes contain no input, URL, header, body, response object, or
exception payload. Use `outcome.contract.name`, status classes, and the bounded
status classification as metric dimensions. Keep full network spans in httpx
or OpenTelemetry instrumentation, with an explicit redaction policy.

Client observers run in declaration order before the call returns or raises.
Their failures are logged and isolated from later observers and from the
caller.

When a call uses an explicit retry policy, pass `attempt_observers=` to receive
one payload-safe `ClientAttemptOutcome` per attempt. It adds the attempt
number, selected retry delay, `will_retry` decision, and the attempt's UTC
`completed_at`. Use the logical outcome for availability and latency; use
attempt outcomes for retry pressure.

## Use bounded labels

Use the contract name, HTTP method, status class, and error source as metric
dimensions. Do not use raw paths, resource IDs, user IDs, idempotency keys, or
error messages as labels; their unbounded values can overwhelm a metrics
backend.

`error_source` distinguishes declared application failures from
framework-owned boundary failures. Record the exact application error code in
structured logs when it is available at the application boundary, but keep
the metric label set controlled.

## Propagate correlation safely

Tenchi accepts a valid inbound `x-request-id` or creates one, exposes it as
`outcome.request.request_id`, and returns it on the response. Configure the
trusted edge so callers cannot inject unrelated tracing headers past your
policy.

When HTTP enqueues work, copy the trusted request or trace correlation into
message metadata. A worker should start or continue a trace from that metadata
and attach its attempt and queue measurements around `dispatcher.dispatch()`. The shared
use-case observer reports the application call; the consumer still owns
transport-specific metadata and acknowledgement.

## Choose the right instrumentation seam

| Need | Use |
| --- | --- |
| Final HTTP status, duration, contract, request ID, error ownership | `RequestOutcome` observer |
| Use-case status and duration across HTTP, jobs, tasks, tools, and `execute()` | `UseCaseOutcome` observer |
| Outbound typed-call status, HTTP status, duration, and declared error code | `ClientOutcome` observer |
| Outbound attempt status, retry decision, and selected delay | `ClientAttemptOutcome` observer |
| Trace the full ASGI lifecycle, unmatched routes, or streaming bodies | ASGI/OpenTelemetry middleware |
| Full database and outbound HTTP spans | Driver or SDK instrumentation |
| Use-case-specific business measurements | An application port called by the use case |
| Worker attempts, queue latency, and correlation | Instrument the consumer around `dispatcher.dispatch()` |

HTTP request and use-case observers run only for matched Tenchi routes. ASGI
middleware is the correct layer for 404s outside the composed route group and
for work that continues while a streaming response body is sent.

## Redact by allowlist

`RequestInfo.headers` is read-only, but it can contain authorization,
cookies, and other credentials. Never log the complete mapping. Select known
safe fields and redact values before serialization.

`ClientOutcome` omits headers and payloads entirely. Instrumentation below the
typed client does not inherit that guarantee.

Apply the same rule to:

- request and response bodies;
- `AppError.details`;
- database statements and bound parameters;
- external service URLs and headers;
- job payloads and dead-letter errors;
- settings and process environment values.

Prefer stable identifiers that help correlate an incident without exposing
personal data. Define retention and access controls for every telemetry sink.

  Logs can be sampled, delayed, redacted, or dropped. A security or compliance
  audit record is application data and needs its own durability, access, and
  retention guarantees.

## Record audit events transactionally

When a business action requires an audit record, define an `AuditLog` port and
write through the same request transaction as the state change:

```python
from typing import Protocol

class AuditLog(Protocol):
    async def record(
        self,
        *,
        actor_id: str,
        action: str,
        subject_id: str,
    ) -> None: ...
```

Use stable action names and the minimum necessary subject identifiers. Do not
store secrets or unrestricted before-and-after payloads. If audit records must
leave the primary database, publish them through the
[transactional outbox pattern](/reliability#commit-deferred-effects-with-state)
after the local record commits.

## Alert on user-visible failure

A useful initial alert set covers:

- elevated framework-owned 5xx responses;
- latency by contract and status class;
- readiness failures;
- database pool exhaustion and transaction errors;
- worker queue age, retry rate, and dead-letter growth;
- OpenAPI or application-tool compatibility gate failures before deployment.

Page on symptoms that require action. Keep lower-level diagnostic events in
logs and traces so an alert links to evidence instead of duplicating it.

---

# Deployment preflight
Source: https://tenchi.io/preflight

`tenchi preflight` answers a different question from `tenchi check`:

| Command | Question | Expected environment |
| --- | --- | --- |
| `tenchi check` | Is this source tree internally valid? | Deterministic and local |
| `tenchi preflight` | Can this release safely use the environment it is about to enter? | The target deployment environment |
| Health and readiness routes | Can this running process serve traffic now? | A live application process |
| `tenchi task run` | Should an authorized operator change application state? | An explicitly selected operational environment |

Use preflight after configuration and migrations are available but before the
new application receives traffic. Typical checks cover database connectivity,
the deployed schema version, secret-manager permissions, required outbound
services, and worker heartbeats.

## Declare read-only observations

The generated application exposes `checks` from
`app/server/preflight.py`. Each observation is a zero-argument async function
that returns `None` on success and raises on failure:

```python
from app.infra.port_wiring import open_preflight_connection
from app.server.runtime import DATABASE_URL
from tenchi.preflight import preflight_check, preflight_group

async def database_connectivity() -> None:
    async with open_preflight_connection(DATABASE_URL) as connection:
        await connection.execute("SELECT 1")

async def database_schema() -> None:
    async with open_preflight_connection(DATABASE_URL) as connection:
        cursor = await connection.execute("PRAGMA user_version")
        row = await cursor.fetchone()
        if row is None or int(row[0]) != 7:
            raise RuntimeError("unexpected database schema")

checks = preflight_group(
    preflight_check(
        "database.connectivity",
        database_connectivity,
        description="Open the configured database in read-only mode.",
        failure_code="DATABASE_UNAVAILABLE",
    ),
    preflight_check(
        "database.schema",
        database_schema,
        description="Verify the schema required by this release.",
        failure_code="DATABASE_SCHEMA_MISMATCH",
    ),
)
```

Names use dotted `snake_case` and remain stable for deployment automation.
Failure codes use `SCREAMING_SNAKE_CASE`. When you omit a code, Tenchi derives
one such as `PREFLIGHT_DATABASE_SCHEMA_FAILED`.

Checks default to a five-second timeout. Set a shorter or longer declaration
timeout only when the dependency has a known response budget:

```python
preflight_check(
    "workers.email",
    email_worker_heartbeat,
    timeout=2.0,
    failure_code="EMAIL_WORKER_NOT_READY",
)
```

Tenchi runs the checks concurrently and reports results in declaration order.
A check must propagate cancellation so its client and connection cleanup can
finish when it times out.

## Keep the boundary read-only

Preflight does not receive an application context and does not start the
application lifespan. Each check must open the narrowest read-only client it
needs. For a database, use a read-only connection or account. For a secret
manager, request metadata or describe a known secret instead of returning its
value. For a worker, read its heartbeat rather than enqueueing probe work.

  Python cannot prove that an arbitrary async function has no side effects.
  Tenchi enforces the narrow function shape, supplies no stateful context,
  marks the MCP tool read-only, and keeps operational mutations in
  `tenchi task run`. Your adapter permissions are the final enforcement
  boundary. Use credentials that cannot write.

Do not migrate schemas, repair records, rotate secrets, enqueue jobs, or call
business endpoints from preflight. Put migrations in the release process and
authorized repairs or backfills in [operational tasks](/tasks).

## Check the production concerns

Keep each dependency independently named so a failed rollout points to one
owner:

- **Database connectivity:** open a read-only connection and run a minimal
  query.
- **Schema compatibility:** read the migration table or required columns and
  compare them with what this release expects.
- **Secret-manager access:** describe required secret references with an
  identity that can read metadata without printing values.
- **Outbound dependencies:** call a documented, non-mutating readiness or
  metadata operation with the production client configuration.
- **Worker readiness:** read a durable heartbeat or consumer-group status and
  reject stale workers.

Preflight should prove only what must be true before traffic shifts. Leave
ongoing dependency monitoring to health checks and observability.

## Run the deployment gate

From the application root:

```shell
uv run tenchi preflight
uv run tenchi preflight --json
uv run tenchi preflight --timeout 3
```

`--timeout` caps every declared timeout; it can make the gate stricter but
never extend a check's own limit. The command exits zero only when every check
passes. Human output names each status and stable failure code. JSON returns
the same versioned result model as the MCP tool:

```json
{
  "schema_version": 4,
  "root": "/srv/api",
  "target": "app.server.preflight:checks",
  "ok": false,
  "counts": {
    "passed": 1,
    "failed": 1,
    "timed_out": 0,
    "total": 2
  },
  "duration_seconds": 0.031,
  "checks": [
    {
      "name": "database.connectivity",
      "description": "Open the configured database in read-only mode.",
      "status": "passed",
      "duration_seconds": 0.012,
      "failure_code": null
    },
    {
      "name": "database.schema",
      "description": "Verify the schema required by this release.",
      "status": "failed",
      "duration_seconds": 0.03,
      "failure_code": "DATABASE_SCHEMA_MISMATCH"
    }
  ]
}
```

Exception types, exception messages, and return values never appear in the
result. Descriptions and failure codes come from the declaration, so keep them
free of credentials and tenant data. A check that accidentally returns a value
is reported as failed and the value is discarded.

  The CLI and MCP tool discard direct stdout and stderr from preflight code.
  They cannot intercept handlers that send logs directly to files, sockets, or
  telemetry backends. Configure dependency logging for production and never
  include secret values in application log records.

Use a different module only when the application cannot follow the generated
convention:

```shell
uv run tenchi preflight --preflight my_app.release:checks
```

For MCP, pass the same target when the server starts:

```shell
uv run tenchi mcp --preflight my_app.release:checks
```

The MCP `preflight` tool is available without enabling task execution. It uses
the credentials and environment captured by the MCP server process, so call it
only when that process is connected to the intended deployment target.

## Place it in the rollout

A safe release sequence is:

1. Validate the source and contracts with `tenchi check` and the OpenAPI and
   application-tool compatibility gates.
2. Load production configuration without printing secret values.
3. Run backward-compatible migrations once.
4. Start the new generation without traffic.
5. Run `tenchi preflight` from that generation's target environment.
6. Wait for live readiness, then shift traffic gradually.
7. Watch request and worker telemetry throughout the rollout.

Continue with [deployment](/deployment) for health routes, server limits,
rollout, and rollback guidance.

---

# Deployment
Source: https://tenchi.io/deployment

Tenchi produces an ordinary ASGI application. Production readiness comes from
explicit lifecycle ownership, middleware, health checks, observability, and a
repeatable contract gate.

Start with the [production handbook](/production) when the application still
needs configuration, transaction, retry, worker, or telemetry decisions. This
page covers the final process and release boundary.

## Run an ASGI server

The generated application exposes `app.server.asgi:app`:

```shell
uv run uvicorn app.server.asgi:app \
  --host 0.0.0.0 \
  --port 8000 \
  --proxy-headers
```

Use your platform's process manager, worker count, graceful shutdown, and
request-timeout guidance. `tenchi dev` enables reload and is not the production
entrypoint.

## Own resources with lifespan

Open database pools, SDK clients, and other process-scoped resources in the
application lifespan. Yield a small state object and build each request context
from it. Cleanup runs during graceful shutdown.

Use a request-scoped context manager for transactions or ports whose success
depends on the request outcome. Application exceptions pass through its
`__aexit__` before Tenchi maps them to HTTP, allowing commit-on-success and
rollback-on-error behavior.

Load [validated configuration](/configuration) before opening resources. Put
database migrations in a controlled release step; use lifespan to open pools
and verify the schema generation that the process expects.

## Add ASGI middleware directly

```python
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware

middleware = [
    Middleware(
        TrustedHostMiddleware,
        allowed_hosts=["api.example.com"],
    ),
    Middleware(
        CORSMiddleware,
        allow_origins=["https://app.example.com"],
        allow_methods=["*"],
        allow_headers=["authorization", "content-type"],
    ),
]

app = create_app(
    routes=routes,
    context_factory=create_context,
    lifespan=lifespan,
    middleware=middleware,
)
```

Configure HTTPS and forwarded headers at the trusted reverse proxy boundary.
Keep host and origin allowlists explicit.

The generated Swagger UI loads pinned CDN assets with subresource integrity.
For an offline deployment, self-host the assets and pass their URLs to
`swagger_ui_route()`. A content-security policy must also allow the page's
fixed inline initializer and style, normally through their hashes. Protect the
UI and `openapi_route()` together when the API description is not public.

## Health and readiness

```python
from tenchi.health import health_route

health = health_route(
    path="/health",
    checks={"database": check_database},
    check_timeout=2.0,
)
```

Asynchronous health checks run concurrently and return `503` when any check is
unhealthy. The route is public by default; pass `public=False` when
infrastructure can authenticate probes.

## Limits and deadlines

The server limits request bodies to 1 MiB by default. Set
`max_request_bytes=` globally or on an individual contract. Declare
`timeout=` for operations that need an application-level deadline, and also
configure transport and platform timeouts.

## Observe outcomes

`RequestInfo` includes a trusted request ID. Tenchi accepts a valid inbound
`x-request-id` or creates one and includes it on responses. Outcome observers
receive the request, final status, duration, and application/framework error
source after request-scoped cleanup.

Use request and use-case observers or ASGI middleware to feed structured logs,
metrics, and traces. They should never contain request bodies, credentials, or
error details unless the application deliberately redacts them.

See [Observability and audit](/observability) for label, correlation, worker,
and redaction conventions.

## Release sequence

Run the local application gate and compare the API with the version currently
deployed:

```shell
uv run tenchi check
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title my_app --diff-ref "$BASE_SHA" --snapshot openapi.json
uv run tenchi tools \
  --diff-ref "$BASE_SHA" --snapshot tools.json
```

`tenchi check` includes the exact OpenAPI and application-tool snapshot checks.
The separate historical compatibility commands remain necessary because they
compare the working contracts with baselines from before the deployment
change.

Then release in this order:

1. Verify production settings and secret references without printing their
   values.
2. Back up data according to the datastore's recovery plan.
3. Run backward-compatible
   [database migrations](/database#run-migrations-as-a-release-step) once.
4. Start the new API generation without sending it traffic.
5. Run [`tenchi preflight`](/preflight) against the target environment.
6. Wait for startup and readiness checks, then shift traffic gradually.
7. Start or update [background workers](/reliability) that are compatible with
   both application generations.
8. Smoke-test a public route and an authenticated operation through the
   production edge.
9. Watch error rate, latency, database saturation, and queue age during the
   rollout.

`tenchi check` does not contact production dependencies, validate secret
manager permissions, run migrations, or prove that an external worker is
healthy. Keep migrations as their own release step, then use
[`tenchi preflight`](/preflight) for read-only, timeout-bounded dependency
checks where the target environment is available.

Plan rollback before rollout. Application rollback must remain compatible with
the migrated schema and any work records already written by the new
generation.

  Raw `httpx.ASGITransport` does not run lifespan by itself. Prefer
  `tenchi.testing.open_client()` or `open_http()`, which start and stop the
  application correctly.

---

# OpenAPI and compatibility
Source: https://tenchi.io/openapi

Tenchi derives OpenAPI 3.1 from composed contracts. The document describes the
same types, headers, errors, media types, visibility, and successful response
definitions enforced by the server and client.

## Generate a document

`openapi_schema()` is a pure function:

```python
from tenchi.openapi import openapi_schema

document = openapi_schema(
    api_routes,
    title="Todos",
    version="1.0.0",
    description="Todo service API",
    security={
        "bearerAuth": {"type": "http", "scheme": "bearer"},
    },
)
```

Pass only the application route group you want documented. Public contracts
are exempted from global security through the same `contract.public` metadata
used by authentication hooks.

## Serve OpenAPI

```python
from tenchi.openapi import openapi_route, swagger_ui_route
from tenchi.routes import route_group

routes = route_group(
    api_routes,
    openapi_route(
        api_routes,
        title="Todos",
        version="1.0.0",
    ),
    swagger_ui_route(title="Todos API"),
)
```

The default path is `/openapi.json`. The serving route is public by default and
does not include itself in the document. Swagger UI defaults to `/docs` and
loads that document through a relative URL, so ASGI `root_path` mounting keeps
both routes aligned.

The default Swagger JavaScript and stylesheet are pinned CDN assets with
subresource integrity metadata. Override `swagger_js_url=`,
`swagger_css_url=`, and `swagger_favicon_url=` to self-host them. Set
`public=False` on both routes when the API description requires authentication.

## Store a canonical snapshot

```shell
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title Todos --write openapi.json
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title Todos --check openapi.json
```

`--write` produces deterministic, key-sorted JSON. `--check` is an exact
equality check suitable for a repository test.

## Classify changes

```shell
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title Todos --diff openapi-baseline.json
```

The analyzer classifies changes as:

- **Additive:** compatible expansion, such as a new optional operation.
- **Metadata:** descriptions and other non-wire changes.
- **Breaking:** removal or tightening that rejects existing valid traffic.
- **Unknown:** JSON Schema behavior that cannot be proven safe.

Breaking and unknown changes return a failing exit status. Use
`--diff-format json` for automation.

  A compatibility gate needs a snapshot from the pull-request base, previous
  push, or previous release. If a breaking change and its new snapshot are
  committed together, comparing the generated document with that same
  snapshot proves only equality.

## CI pattern

```shell
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title Todos --diff-ref "$BASE_SHA" --snapshot openapi.json
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title Todos --check openapi.json
```

`--diff-ref` resolves the snapshot inside the current Git repository and reads
`REF:PATH` without changing the working tree. Missing refs and snapshots fail
the command. Generated CI uses the pull request's base SHA; push checks retain
exact snapshot validation without assuming a previous commit exists.

Use identical routes, title, version, description, and security options for
generation, comparison, and snapshot checks.

Programmatic consumers can import `analyze_openapi_compatibility()` from
`tenchi.compatibility` and inspect its structured `CompatibilityReport`.

Contracts marked `webhook=True` include `x-tenchi-webhook: true` on their
OpenAPI operation. Adding that requirement is breaking because existing
callers must begin signing deliveries; removing it is additive.

---

# CLI
Source: https://tenchi.io/cli

The `tenchi` command scaffolds the prescribed structure, inspects the composed
application, and keeps API changes reviewable.

## Create an application

```shell
uvx tenchi new my_app
```

Names use `snake_case`. The generated project includes a todos feature,
SQLite persistence with request-scoped transactions, a memory test adapter,
strict checks, integration tests, Swagger UI, OpenAPI and application-tool
snapshots, a concise `AGENTS.md`, project-local
MCP configuration, and GitHub Actions CI.

## Generate application slices

```shell
uv run tenchi make feature notes
uv run tenchi make feature notes --dry-run
uv run tenchi make use-case notes create_note
uv run tenchi make use-case notes create_note --json
```

Generators create files and print explicit wiring instructions. They never
rewrite route or infrastructure modules. `--dry-run` validates the operation
and lists every file without writing it. `--json` emits the same result as a
versioned object with the app root, artifact identity, files, follow-up steps,
and any error.

## Inspect routes

```shell
uv run tenchi routes
uv run tenchi routes --json
```

The default target is `app.server.routes:routes`. Override it with
`--routes module:attribute`. JSON output is a versioned object containing the
application root and the composed HTTP surface under `routes`.

## Map the application

```shell
uv run tenchi map
uv run tenchi map --json
uv run tenchi map --feature notes --json
uv run tenchi map --feature notes \
  --kind route,job,task,tool,use-case,policy,port --json
```

Map combines the canonical source layout with composed API routes, operational
tasks, background jobs, and application tools. It returns a deterministic,
versioned graph covering features, contracts, routes, background jobs,
operational tasks, application tools, use cases, policies, ports, adapters,
context types, entrypoints, and tests.
Edges describe ownership, route bindings, dependencies, authorization,
implementations, and feature tests. Each edge carries project-relative source
evidence and an `exact` or `inferred` confidence value.

`--feature` keeps the selected feature and its directly connected nodes, which
makes cross-feature policy and shared-port dependencies visible without loading
the entire application. `--kind` accepts a comma-separated projection of node
kinds. The default route target is `app.server.routes:api_routes`; override it
with `--routes module:attribute`. The default job target is
`app.server.jobs:jobs`; override it with `--jobs module:attribute`. The default
tool target is `app.server.tools:tools`; override it with
`--tools module:attribute`.

The JSON result also embeds `tenchi doctor` diagnostics and unresolved source
relationships. Agents should inspect both before editing and use stable node
IDs and source locations to choose the files involved in a change. See [coding
agents](/agents) for the complete workflow across map, make, check, and OpenAPI
compatibility.

## Manage application-tool contracts

```shell
uv run tenchi tools
uv run tenchi tools --json
uv run tenchi tools --diff tools.json
uv run tenchi tools \
  --diff-ref origin/main \
  --snapshot tools.json
uv run tenchi tools --check tools.json
uv run tenchi tools --write tools.json
```

The default target is `app.server.tools:tools`; override it with
`--tools module:attribute`. Plain output is the canonical portable manifest.
`--json` wraps that manifest with the application root and agent protocol
version for automation.

`--check` is an exact drift check. `--diff` and `--diff-ref` classify changes
directionally and fail on breaking or unknown changes. A Git ref supplies a
meaningful historical baseline even when the working branch updates its
snapshot. Use `--diff-format json` for the versioned compatibility result. See
[Application tools](/tools) for the compatibility rules.

## Manage OpenAPI

```shell
uv run tenchi openapi --routes app.server.routes:api_routes --title my_app
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title my_app --diff openapi.json
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title my_app --diff-ref origin/main --snapshot openapi.json
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title my_app --check openapi.json
uv run tenchi openapi --routes app.server.routes:api_routes \
  --title my_app --write openapi.json
```

Common metadata options are `--title`, `--version`, `--description`, and
`--security`. `--diff-ref` reads `--snapshot` (default `openapi.json`) from a
Git commit instead of the working tree. See [OpenAPI and
compatibility](/openapi) for the safe baseline workflow.

When `--diff-format json` is selected, the versioned result includes the
application root, baseline label, compatibility status, severity counts, and
classified changes.

## Check architecture

```shell
uv run tenchi doctor
uv run tenchi doctor --json
```

Doctor validates the canonical application structure, dependency direction,
and authorization consistency. Findings include a stable code, severity, file,
line, and message in the versioned JSON result.

## Run every check

```shell
uv run tenchi check
uv run tenchi check --json
```

Check runs Ruff format, Ruff lint, Pyright, pytest, doctor, and the exact
OpenAPI and application-tool snapshot checks. Every step runs even when an
earlier one fails. Human output shows a compact status list; JSON includes
stable step names, commands, exit codes, durations, and bounded failure output.
Use `--timeout` to change the per-step limit.

OpenAPI defaults come from literal top-level `OPENAPI_TITLE`, `OPENAPI_VERSION`,
optional `OPENAPI_DESCRIPTION`, and optional `OPENAPI_SECURITY` declarations in
the module selected by `--routes`; command flags override them. The route target
defaults to `app.server.routes:api_routes`, and the snapshot defaults to
`openapi.json`. The tool target defaults to `app.server.tools:tools`, and its
snapshot defaults to `tools.json`.

## Verify a completed change

```shell
uv run tenchi verify --base-ref origin/main
uv run tenchi verify --base-ref origin/main --json
```

Verify produces one receipt for the finished source tree. It runs `tenchi
check`, rejects application-map diagnostics and unresolved relationships, and
compares the generated OpenAPI document and application-tool manifest with the
snapshots at the selected Git ref. The receipt records the immutable commit
resolved from `--base-ref`, so both compatibility reports use the same
historical state even if the named ref moves later.

Use the pull request base, previous push, or previous release as the base ref.
The option is required: using the current branch snapshot could hide a breaking
change when code and its updated snapshot are committed together. The command
exits non-zero for a failed check, incomplete architecture evidence, a breaking
or unknown boundary change, or a baseline that cannot be read.

`--snapshot`, `--tool-snapshot`, and the route, task, job, tool, and OpenAPI
metadata options override the generated application conventions. Verify never
updates either snapshot. Because it runs the application's tests and validation
commands, those commands retain their normal side effects and per-step timeout.

## Verify the deployment environment

```shell
uv run tenchi preflight
uv run tenchi preflight --json
uv run tenchi preflight --timeout 3
```

Preflight discovers `app.server.preflight:checks` and runs its read-only async
observations concurrently. Each check keeps its declared timeout; `--timeout`
can only cap those limits. Results expose stable names, descriptions, statuses,
durations, and failure codes while discarding dependency values and exception
messages. The command exits non-zero when any check fails or times out.

Use `--preflight module:attribute` to override the declaration target. See
[deployment preflight](/preflight) for declarations, dependency patterns,
redaction, and rollout placement.

## Run operational tasks

```shell
uv run tenchi task list
uv run tenchi task list --json
uv run tenchi task run projects.repair_members \
  --input '{"dry_run": true}'
uv run tenchi task run projects.repair_members \
  --input '{"dry_run": false}' \
  --json
```

`task list` discovers `app.server.tasks:runner` and reports every task's input
and output JSON Schema. `task run` validates input, opens the application
lifespan and scoped context, invokes the use case, validates its result before
the context commits, and returns a non-zero exit status for failure results.
Use `--tasks module:attribute` to override the runner target.

See [Operational tasks](/tasks) for declarations, composition, dry-run design,
failure semantics, and MCP access.

## Serve coding-agent tools over MCP

```shell
uv run tenchi mcp
uv run tenchi mcp --root path/to/application
```

This command exposes repository inspection and validation—not the
application's `ToolGroup`. MCP support is installed through the `tenchi[mcp]`
extra. Generated
applications include it as a development dependency and register the default
command in `.mcp.json`.
`--routes`, `--api-routes`, `--preflight`, `--tasks`, `--jobs`, `--tools`,
`--snapshot`, and `--tool-snapshot` override the conventions captured by the
server when it starts.
`--title`, `--version`,
`--description`, and `--security` override discovered OpenAPI metadata for the
diff and check tools. Pass `--allow-task-runs` only when the connected agent
may perform operational writes; task discovery remains available without it.
The command uses stdio and waits for an MCP client; it does not start an HTTP
listener.

See [connect an MCP-aware coding agent](/mcp) for tool behavior, safety, and
client configuration.

## Run development

```shell
uv run tenchi dev
uv run tenchi dev --host 0.0.0.0 --port 8080
uv run tenchi dev --no-reload
```

The default ASGI target is `app.server.asgi:app`. Override it with
`--app module:attribute`. Production deployments should invoke an ASGI server
directly rather than the development command.

---

# Coding agents
Source: https://tenchi.io/agents

Tenchi gives coding agents deterministic inspection, previewable generation,
structured diagnostics, and one complete validation command. An MCP-aware
agent can access those operations directly through the [Tenchi MCP
server](/mcp); an agent with filesystem and shell access can use the equivalent
CLI workflow.

You remain in control because contracts, use cases, ports, and composition stay
in ordinary Python files. The agent inspects evidence, edits those files
directly, and runs the same checks you use locally and in CI.

## Give an agent a Tenchi task

Generated applications include an `AGENTS.md`, so most agents will discover
the local rules automatically. For an explicit first instruction, adapt this
prompt:

```text
Read AGENTS.md before editing. Run `uv run tenchi map` for the affected feature
and inspect its diagnostics and unresolved references. Explain the intended
files and relationships before changing them. Preview any generated structure
with `uv run tenchi make --dry-run`. Implement the change without hiding
explicit wiring, then run `uv run tenchi check`. If a contract changed, run the
OpenAPI compatibility diff before updating its snapshot. If an application
tool changed, run the tool compatibility diff before updating `tools.json`.
Finish with `uv run tenchi verify --base-ref <historical-ref> --json` and report
the receipt with the files changed.
```

If you already know the feature, include its name in the task so the agent can
start with `uv run tenchi map --feature  --json` instead of loading the
complete application graph.

## What Tenchi provides

| Design choice | What it gives an agent |
| --- | --- |
| Canonical application structure | Predictable locations for contracts, behavior, ports, policies, adapters, and composition |
| `AGENTS.md` in every generated app | Repository-local placement rules, dependency direction, and a validation loop |
| `.mcp.json` in every generated app | Project-local registration for Tenchi's MCP tools and instructions resource |
| `tenchi map --json` | A versioned graph with stable node IDs, source locations, registration state, and relationship evidence |
| `tenchi task list --json` | Validated operational task names and their input/output JSON Schemas |
| `tenchi make --dry-run --json` | A mutation preview with the files and follow-up wiring steps before anything is written |
| `tenchi doctor --json` | Stable diagnostic codes and source locations for architectural violations |
| `tenchi tools --json` | Registered machine-facing names, schemas, errors, and safety annotations |
| `tenchi check --json` | One bounded, complete result for formatting, linting, types, tests, architecture, and boundary snapshot drift |
| `tenchi verify --base-ref  --json` | One completion receipt tied to an immutable commit, including checks, strict architecture evidence, and both compatibility reports |
| `tenchi preflight --json` | Redacted, timeout-bounded evidence that the selected deployment environment is ready |
| OpenAPI compatibility commands | A historical contract baseline that an agent cannot accidentally replace and then compare to itself |
| Tool compatibility commands | Directional proof that machine-facing input, output, errors, and safety did not break existing callers |

Tenchi's documentation is also published as [/llms.txt](/llms.txt) for compact
navigation and [/llms-full.txt](/llms-full.txt) for the complete guide.

## Use the recommended working loop

### 1. Read the local rules

Start with the generated `AGENTS.md`. It describes the application's file
layout, allowed dependency direction, explicit composition points, error and
authorization conventions, and the commands that define done.

Tell the agent to follow those repository instructions ahead of generic
framework advice; your application may add stricter local conventions.

### 2. Map before reading broadly

```shell
uv run tenchi map --json
uv run tenchi map --feature notes --json
uv run tenchi map --feature notes \
  --kind contract,route,use-case,policy,port,adapter --json
```

The complete map combines source declarations with composed API routes,
operational tasks, background jobs, and application tools. A feature projection
retains directly connected shared and cross-feature nodes, giving an agent a
bounded starting context without hiding dependencies that cross the feature
directory.

Inspect these fields before editing:

- `nodes` identify concepts and their source declarations. Runtime-bound kinds
  such as contracts, routes, jobs, tasks, tools, use cases, and adapters become
  `registered` when composition reaches them; source-only declarations remain
  `declared`.
- `edges` explain ownership, route bindings, dependencies, authorization,
  implementations, and feature tests. Evidence points to the source that
  supports each relationship.
- `confidence` is `exact` when the relationship is directly proven and
  `inferred` when it follows a documented naming convention.
- `diagnostics` embeds the current doctor findings.
- `unresolved` records relationships the analyzer could not prove. Treat these
  as missing context, not as permission to assume the dependency is absent.

Node IDs, result keys, and diagnostic codes are stable within the declared
`schema_version`. If your tooling parses the JSON directly, check that version
before relying on the rest of the result. Tenchi snapshots the JSON Schema for
every structured CLI result and MCP tool input and output in CI. Additive
changes may retain the current version; breaking or unknown changes require a
new version and preserve the older baseline.

### 3. Preview generated structure

```shell
uv run tenchi make feature notes --dry-run --json
uv run tenchi make use-case notes create_note --dry-run --json
```

Dry runs perform the same naming and conflict validation as a real generation
without writing files. The result lists every planned path and the explicit
wiring steps that remain app-owned.

After accepting the preview, run the command without `--dry-run`. Generators
create files but do not rewrite route or infrastructure modules. The agent must
make those composition changes in the application files, where you can review
them.

### 4. Edit through the application boundaries

Follow the [application architecture](/architecture) rather than placing code
where it is merely convenient:

- Contracts own the HTTP boundary.
- Plain async use cases own behavior.
- Pure policies own authorization decisions.
- Feature-owned protocols define infrastructure needs.
- Adapters implement those ports under `app/infra/`.
- Server modules explicitly compose routes, contexts, hooks, and concrete
  adapters.

If the graph and the task disagree, inspect the source evidence and unresolved
references before expanding the edit. `tenchi map` is an orientation tool, not
a substitute for reading the definitions that will change.

### 5. Validate once, completely

```shell
uv run tenchi check --json
```

Check runs every step even after one fails, so you receive a complete repair
list in one pass. Each step reports its command, exit code, duration, bounded
standard output and error output, and whether either stream was truncated.

Treat `ok: false` as unfinished work. Do not stop after fixing only the first
failed step, and do not silently weaken Ruff, Pyright, pytest, doctor, or the
OpenAPI and application-tool snapshots to make the aggregate pass.

### 6. Compare boundary changes historically

For a contract change, compare before replacing the snapshot:

```shell
uv run tenchi openapi \
  --routes app.server.routes:api_routes \
  --title my_app \
  --diff openapi.json

uv run tenchi openapi \
  --routes app.server.routes:api_routes \
  --title my_app \
  --diff-ref origin/main \
  --snapshot openapi.json
```

Breaking and unknown changes fail closed. A baseline from the merge base,
previous push, or previous release remains meaningful even when the branch also
updates its committed snapshot. See [OpenAPI and compatibility](/openapi) for
the complete workflow.

For an application-tool change, use the parallel workflow:

```shell
uv run tenchi tools --diff tools.json
uv run tenchi tools \
  --diff-ref origin/main \
  --snapshot tools.json
```

This catches removed tools, narrower inputs, wider outputs, newly possible
application errors, and less-safe annotations before an agent replaces the
evidence. See [Application tools](/tools) for the complete compatibility rules.

  Do not use `--write` before reviewing `--diff`. Comparing the generated
  document or manifest with the snapshot updated in the same change proves
  equality, not compatibility.

### 7. Produce the completion receipt

After reviewing and accepting any snapshot updates, verify the finished tree
against the same historical point:

```shell
uv run tenchi verify --base-ref origin/main --json
```

The receipt contains the resolved commit, complete check result, application
summary, diagnostics, unresolved relationships, and both compatibility
reports. It passes only when every check passes, the map has no diagnostics or
unresolved relationships, and neither boundary contains a breaking or unknown
change.

`verify` does not replace the diff-before-write review. It proves that the
finished tree—including accepted snapshot updates—still agrees with the
historical contract. Use the pull request base, previous push, or previous
release instead of the branch's current commit.

## Choose the right command

| Command | Structured result |
| --- | --- |
| `tenchi map --json` | Root, summary, nodes, edges, diagnostics, and unresolved references |
| `tenchi make ... --json` | Artifact identity, dry-run state, files, next steps, and structured failure |
| `tenchi doctor --json` | Overall status and source-anchored diagnostics with stable codes |
| `tenchi check --json` | Overall status, counts, duration, and every validation step |
| `tenchi verify --base-ref  --json` | Resolved baseline commit, checks, strict architecture evidence, and OpenAPI and tool compatibility |
| `tenchi preflight --json` | Deployment-environment status, counts, durations, and stable failure codes without dependency details |
| `tenchi routes --json` | A versioned result containing the app root and composed HTTP route table |
| `tenchi tools --json` | A versioned result containing the app root and registered tool manifest |
| `tenchi openapi --diff openapi.json --diff-format json` | A machine-readable API compatibility report |
| `tenchi tools --diff tools.json --diff-format json` | A machine-readable tool compatibility report |

Human and JSON modes represent the same operation. Use JSON when another tool
will make a decision; use human output when a person is reviewing the work in a
terminal.

## Know when to inspect the source yourself

The application map is source-backed and conservative. Python permits dynamic
imports, factories, decorators, and runtime mutation that static analysis may
not be able to prove. Tenchi reports unresolved relationships rather than
inventing certainty, and inferred edges remain visibly different from exact
ones.

The map also does not prove business correctness. An agent still needs to read
the relevant contract, use case, policy, port, and tests, then rely on the full
validation loop.

  Generated applications include `.mcp.json` for MCP-aware clients. Agents that
  only have shell access can follow the same workflow through the CLI and
  generated `AGENTS.md`; the result schemas and validation semantics stay the
  same.

Continue with the [coding-agent MCP guide](/mcp) to connect a client, the [CLI
reference](/cli) for every command option, or [stability and
releases](/stability) for the compatibility guarantees around these result
schemas.

---

# Connect a coding agent over MCP
Source: https://tenchi.io/mcp

Tenchi's MCP server gives a coding agent structured access to the application
map, route table, registered application-tool contract, architecture
diagnostics, generator previews, OpenAPI and tool compatibility, deployment
preflight, operational-task discovery, and the complete validation loop.
Inspection and preview tools do not edit application files; `check` runs the
project's own validation commands and can have their usual side effects.

  `tenchi mcp` exposes repository inspection and validation to a coding agent.
  To expose your application's own `ToolGroup` to users or AI features, see
  [Serve application tools over MCP](/tool-mcp).

New applications include the required development dependency and a
project-local `.mcp.json`. After `uv sync`, an MCP client that recognizes this
file can start the server with no additional Tenchi configuration.

## Add MCP to an existing application

Install the optional dependency:

```shell
uv add --dev "tenchi[mcp]"
```

Create `.mcp.json` at the application root:

```json
{
  "mcpServers": {
    "tenchi": {
      "command": "uv",
      "args": ["run", "tenchi", "mcp", "--root", "."]
    }
  }
}
```

If your client does not read project-local MCP configuration, register the
same command and arguments through that client's MCP settings. The server uses
stdio; starting `tenchi mcp` directly leaves it waiting for an MCP client on
standard input.

## Give the agent project context

The `tenchi://project/agents` resource contains the application's `AGENTS.md`.
Ask the agent to read it before changing code. If the file is absent, the
resource supplies a short fallback workflow and a link to the full coding-agent
guide.

The server captures one application root when it starts. Every tool operates
inside that root, and OpenAPI and application-tool snapshot paths cannot escape
it. Inspection tools reload the application's source for each call, so a server
that stays open sees edits made during the agent session.

## Use the tools

| Tool | Result |
| --- | --- |
| `app_map` | Versioned nodes, relationships, evidence, diagnostics, and unresolved references; accepts feature and node-kind projections |
| `routes` | The composed route table with use cases, responses, errors, access metadata, and runtime limits |
| `tools` | The registered application-tool manifest with input/output schemas, declared errors, and safety annotations |
| `doctor` | Source-anchored architecture and dependency diagnostics |
| `preflight` | Read-only, timeout-bounded observations of the target deployment environment with redacted results |
| `task_list` | Registered operational tasks with validated input and output JSON Schemas |
| `make_preview` | The files and wiring steps for a feature or use case, always with `dry_run: true` |
| `openapi_diff` | A compatibility report against `openapi.json`, another project snapshot, or that snapshot at a Git ref |
| `tools_diff` | A directional compatibility report against `tools.json`, another project snapshot, or that snapshot at a Git ref |
| `verify` | One completion receipt containing checks, strict architecture evidence, and both compatibility reports against a required Git ref |
| `check` | Ruff format, Ruff lint, Pyright, pytest, doctor, and the OpenAPI and tool snapshot checks |

`task_run` is absent by default. Start the server with
`--allow-task-runs` to expose it:

```shell
uv run tenchi mcp --allow-task-runs
```

The tool can change application state and uses the credentials available to
the MCP server process. Only enable it for an agent and environment authorized
to perform operational work. See [Operational tasks](/tasks) for the complete
workflow.

`preflight` is available by default and is marked read-only. It still contacts
the environment selected by the MCP server process. Call it only when that
process has the intended deployment configuration and read-only dependency
credentials. See [deployment preflight](/preflight) for the application-side
contract and redaction boundary.

Every tool returns structured content with a `schema_version`. A doctor
finding, failed check, generator conflict, or incompatible API or tool contract
is a valid result with `ok: false` or `compatible: false`; malformed arguments
and unreadable or unsafe paths are MCP errors. Tenchi snapshots every tool's
input and output schema in CI; breaking or unknown changes require a new
protocol version.

  `check` and `verify` execute the application's tests and validation commands.
  Those commands can have their usual filesystem, database, or network side
  effects. MCP clients should treat them as actions rather than read-only
  inspection. Cancelling either tool stops the active validation process.

## Follow the agent loop

For a feature change, give the agent this sequence:

1. Read `tenchi://project/agents`.
2. Call `app_map` with the feature name and inspect diagnostics and unresolved
   relationships.
3. Call `make_preview` when new framework-shaped files are needed.
4. Edit ordinary Python files through the agent's normal filesystem tools.
5. Call `check` and resolve every failed step.
6. Call `openapi_diff` before accepting a changed OpenAPI snapshot.
7. Call `tools_diff` before accepting a changed application-tool snapshot.
8. Call `verify` with the pull request base, previous push, or previous release
   and retain its completion receipt.

The agent must still make route and infrastructure wiring explicit in the
application source. MCP previews follow the same rule as `tenchi make`: they
describe the remaining wiring instead of silently changing composition files.

## Override application conventions

The generated structure works with the defaults. For an application using
different module targets or snapshot location, change the registered command:

```shell
uv run tenchi mcp \
  --root . \
  --routes my_app.server.routes:routes \
  --api-routes my_app.server.routes:api_routes \
  --preflight my_app.server.preflight:checks \
  --tasks my_app.server.tasks:runner \
  --jobs my_app.server.jobs:jobs \
  --tools my_app.server.tools:tools \
  --snapshot api/openapi.json \
  --tool-snapshot api/tools.json \
  --title "My API" \
  --version 1.0.0
```

`routes` uses `--routes`. The map, OpenAPI diff, `check`, and `verify` tools use
`--api-routes`; `preflight` uses `--preflight`; the task tools use `--tasks`.
The map loads registered background jobs through `--jobs` and registered
application tools through `--tools`. Tool discovery and compatibility also use
`--tools`; `tools_diff`, `check`, and `verify` use `--tool-snapshot`.
OpenAPI title, version,
description, and security defaults come from the same literal route-module
declarations used by `tenchi check`.
`--title`, `--version`, `--description`, and `--security` provide the same
explicit overrides when an application does not keep that metadata as literals.

Continue with [coding agents](/agents) for the complete source-editing workflow
or the [CLI reference](/cli) when an agent can run shell commands but not MCP.

---

# Module reference
Source: https://tenchi.io/reference

Tenchi's supported Python modules each own one small responsibility. Import
from the module that owns a declaration when you want its full API, or use the
selected package-root re-exports for compact application code.

## Package-root imports

`tenchi` re-exports the declarations most applications use:

```python
from tenchi import (
    IDEMPOTENCY_CONFLICT,
    IDEMPOTENCY_IN_PROGRESS,
    TOOL_MANIFEST_VERSION,
    AppError,
    Client,
    ClientAttemptObserver,
    ClientAttemptOutcome,
    ClientObserver,
    ClientOutcome,
    ErrorDef,
    IdempotencyResultError,
    IdempotencyStore,
    IdempotencyStoreError,
    Job,
    JobDispatcher,
    JobMessage,
    Page,
    PageQuery,
    PreflightGroup,
    RATE_LIMITED,
    MemoryRateLimitStore,
    RateLimitStore,
    RetryPolicy,
    RetryTimeoutError,
    Tool,
    ToolRunner,
    UseCaseObserver,
    UseCaseOutcome,
    Webhook,
    WebhookRequest,
    contract,
    create_app,
    create_job_dispatcher,
    create_task_runner,
    create_tool_runner,
    execute,
    enforce_rate_limit,
    fingerprint,
    health_route,
    job,
    job_group,
    job_handler,
    job_message,
    openapi_route,
    page,
    preflight_check,
    preflight_group,
    present,
    response,
    retry_policy,
    route,
    route_group,
    run_idempotently,
    swagger_ui_route,
    task,
    task_group,
    tool,
    tool_group,
    tool_handler,
    tool_manifest,
    webhook,
)
```

Canonical examples use the owning submodule when that makes the responsibility
clearer.

## Core declarations

| Module | Public names |
| --- | --- |
| `tenchi.contracts` | `Contract`, `contract()` |
| `tenchi.routes` | `Route`, `RouteGroup`, `RouteBindingError`, `UseCase`, `route()`, `route_group()` |
| `tenchi.responses` | `ResponseDef`, `PresentedResponse`, `response()`, `present()` |
| `tenchi.errors` | `ErrorDef`, `AppError`, `TenchiError`, `ConfigurationError`, `ERROR_SOURCE_HEADER`, `REQUEST_ID_HEADER`, `error_body()` |

Contracts describe HTTP. Routes bind contracts to use cases. Response
definitions describe successful wire outcomes. Error definitions describe
expected application failures.

## Runtime

| Module | Public names |
| --- | --- |
| `tenchi.server` | `create_app()`, `RequestInfo`, `RequestOutcome`, `Hook`, `OutcomeObserver`, `ContextFactory`, `Lifespan`, `DEFAULT_MAX_REQUEST_BYTES`, `ERROR_SOURCE_HEADER`, `REQUEST_ID_HEADER` |
| `tenchi.execution` | `execute()`, `open_context()`, `ExecutionError`, `UseCaseOutcome`, `UseCaseObserver` |
| `tenchi.idempotency` | `IdempotencyStore`, `IdempotencyReservation`, `IdempotencyReplay`, `IdempotencyConflict`, `IdempotencyInProgress`, `IdempotencyDecision`, `IdempotencyResultError`, `IdempotencyStoreError`, `MemoryIdempotencyStore`, `IDEMPOTENCY_CONFLICT`, `IDEMPOTENCY_IN_PROGRESS`, `fingerprint()`, `run_idempotently()` |
| `tenchi.jobs` | `Job`, `JobMessage`, `JobHandler`, `JobGroup`, `JobDispatcher`, `JobBindingError`, `JobNotFoundError`, `JobResultError`, `job()`, `job_message()`, `job_handler()`, `job_group()`, `create_job_dispatcher()` |
| `tenchi.preflight` | `PreflightCheck`, `PreflightGroup`, `PreflightOutcome`, `PreflightReport`, `PreflightStatus`, `PreflightBindingError`, `preflight_check()`, `preflight_group()`, `run_preflight()` |
| `tenchi.rate_limits` | `RateLimitStore`, `RateLimitPermit`, `RateLimitExceeded`, `RateLimitDecision`, `RateLimitStoreError`, `MemoryRateLimitStore`, `RATE_LIMITED`, `enforce_rate_limit()` |
| `tenchi.tasks` | `Task`, `TaskGroup`, `TaskRunner`, `task()`, `task_group()`, `create_task_runner()` |
| `tenchi.tools` | `TOOL_MANIFEST_VERSION`, `Tool`, `ToolBinding`, `ToolGroup`, `ToolRunner`, `ToolBindingError`, `ToolNotFoundError`, `ToolResultError`, `ToolInvocationError`, `tool()`, `tool_handler()`, `tool_group()`, `create_tool_runner()`, `tool_manifest()` |
| `tenchi.webhooks` | `Webhook`, `WebhookRequest`, `WebhookVerifier`, `WebhookBindingError`, `webhook()` |
| `tenchi.client` | `Client`, `ClientResponse`, `ClientOutcome`, `ClientObserver`, `ClientAttemptOutcome`, `ClientAttemptObserver`, `UnexpectedResponseError` |
| `tenchi.retries` | `RetryPolicy`, `RetryTimeoutError`, `retry_policy()` |
| `tenchi.testing` | `open_client()`, `open_http()`, `verify_idempotency_store()`, `verify_rate_limit_store()`, `StoreConformanceError`, `IdempotencyStoreFactory`, `RateLimitStoreFactory`, `ClockAdvance`, `ASGIApp` |

`create_app()` owns HTTP dispatch and lifecycle composition. `execute()` runs
the application boundary without HTTP. A preflight group defines read-only,
timeout-bounded observations for a deployment gate. A `TaskRunner` adds stable
discovery, input and result validation, and application lifecycle wiring for
operational commands. `Client` enforces contracts against remote or in-process
servers; an explicit `RetryPolicy` coordinates bounded logical retries. A
`JobDispatcher` validates durable messages and their registered consumer
results without owning a queue. A `Webhook` binds an exact-body verifier to a
contract marked `webhook=True`. A `RateLimitStore` atomically applies
fixed-window application quotas. An application `ToolRunner` validates
machine-facing calls, applies lifecycle and context wiring, and masks
undeclared failures. `tool_manifest()` exposes portable JSON Schemas and
safety metadata without choosing an agent transport.

## Optional integrations

| Module | Public names | Install |
| --- | --- | --- |
| `tenchi.mcp` | `TOOL_MCP_PROTOCOL_VERSION`, `McpRequest`, `create_tool_mcp_server()` | `uv add "tenchi[mcp]"` |
| `tenchi.opentelemetry` | `OpenTelemetryObservers`, `create_opentelemetry_observers()` | `uv add "tenchi[otel]"` |

The MCP module exposes an authenticated `ToolGroup` through the MCP SDK while
the application owns identity, visibility, approval policy, and runner wiring.
The OpenTelemetry module records through application-configured providers. It
does not create an SDK, exporter, background worker, or shutdown lifecycle.

## Schema and operations

| Module | Public names |
| --- | --- |
| `tenchi.openapi` | `openapi_schema()`, `openapi_route()`, `swagger_ui_route()` |
| `tenchi.compatibility` | `CompatibilityChange`, `CompatibilityReport`, `analyze_openapi_compatibility()`, `analyze_tool_compatibility()`, and their report renderers |
| `tenchi.pagination` | `Page`, `PageQuery`, `page()` |
| `tenchi.health` | `HealthCheck`, `HealthReport`, `health_route()` |

Use the [`tenchi` CLI](/cli) for scaffolding, application inspection,
architecture diagnostics, project checks, verification receipts, and OpenAPI
and application-tool snapshots. Modules such as `tenchi.cli`,
`tenchi.doctor`, `tenchi.scaffold`, and `tenchi.snapshots` implement that
command surface; they are not supported application imports.

## Exact signatures

Tenchi ships inline type information and is checked with strict Pyright. Use
editor completion for overloads and parameter types, or inspect the supported
module source on
[GitHub](https://github.com/taylorbryant/tenchi/tree/main/src/tenchi).

Pre-1.0 releases may change these APIs between minor versions. Follow the
[safe upgrade workflow](/stability) before updating an application.

---

# Stability and releases
Source: https://tenchi.io/stability

Tenchi is pre-1.0. It follows semantic versioning with pre-1.0 semantics: minor
releases may change public APIs while the framework finds the cleanest durable
shape.

## What is protected today

Every change on `main` runs formatting, linting, strict Pyright, framework
tests, standalone application tests, architecture checks, exact OpenAPI
and application-tool snapshots, and compatibility analysis against historical
baselines.

The repository also snapshots public module signatures. An accidental API
change fails CI; an intentional one must update the snapshot and changelog.

Agent-facing JSON has explicit compatibility boundaries. Structured CLI results
and the coding-agent MCP server share one versioned schema, and CI verifies
that their schemas remain identical. Application-tool manifests and the
application MCP adapter each use their own version and retained schema
snapshots. Breaking or unknown changes require a new versioned snapshot while
the previous baseline remains unchanged.

These checks make change deliberate. They do not promise that every pre-1.0
release is source-compatible.

## Upgrade safely

1. Read the release entry in `CHANGELOG.md`.
2. Upgrade Tenchi in a branch and refresh the lockfile.
3. Run `tenchi check` to verify formatting, lint, types, tests, architecture,
   and the current OpenAPI and application-tool snapshots.
4. Run `tenchi openapi --diff` against the pre-upgrade snapshot.
5. Run `tenchi tools --diff` against the pre-upgrade snapshot.
6. Review application wiring and generated-code convention changes.
7. Update snapshots only after accepting both compatibility reports.
8. Run `tenchi verify --base-ref ` to produce one final
   receipt against the pre-upgrade commit.

Pin a compatible minor range or an exact version according to the application's
risk tolerance. Production applications should not upgrade framework versions
implicitly during deployment.

## Deprecation metadata

Contract deprecation is part of the API itself. Use `deprecated=True` while an
operation remains available, or provide a timestamp. Add `sunset=` when callers
need a concrete removal date. Tenchi emits the lifecycle metadata in OpenAPI
and response headers.

## What will gate 1.0

Tenchi should reach 1.0 after its contract, response, client, context, and
composition APIs have been exercised by several independently maintained
applications and can remain stable across normal framework evolution.

The bar is not a large feature checklist. It is confidence that the small core
has earned its abstractions and that users can upgrade predictably.

  Bugs and design feedback from complete applications are more valuable than
  speculative feature parity. Open an issue with the contract, use-case shape,
  and boundary behavior that created the pressure.