# Tenchi
Source: https://tenchi.io
Tenchi
Build typed Python APIs around explicit contracts and plain async functions.
Tenchi validates data at the boundary, keeps behavior independent of HTTP, and
leaves infrastructure choices to your application.
## Choose a path
## Start with a running API
```shell
uvx tenchi new my_app
cd my_app
uv sync
uv run tenchi dev
```
Call the generated todos API from another terminal:
```shell
curl -i \
-H 'content-type: application/json' \
-d '{"title":"Buy milk"}' \
http://127.0.0.1:8000/todos
```
The generated application includes a working feature, SQLite persistence,
direct use-case tests, HTTP tests, and Swagger UI at
[http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs).
## One small application model
```text
validated input -> contract + route -> async use case -> app-owned port -> adapter
```
- A **contract** declares an HTTP operation and its validated data.
- A **route** binds that contract to a plain async use case.
- The **use case** owns behavior without depending on Starlette or a database
implementation.
- An application-owned **context** supplies ports and verified identity.
- **Adapters** implement those ports for SQLite, another database, an external
API, or a focused test.
Tenchi checks the route and use-case signature when the application is
composed. At runtime, it validates input before the use case and validates the
result before the request scope commits.
Read [How Tenchi works](/concepts) for the complete mental model or
[Build a feature](/build-a-feature) to carry a new operation through
persistence and tests.
## Add capabilities when the application needs them
The same use-case model supports more than HTTP, but those capabilities are
optional:
- [Build the core API](/contracts) with declared errors, authentication, pagination,
OpenAPI, and a typed Python client.
- [Prepare for production](/production) with explicit transactions,
idempotency, retries, observability, preflight, and deployment decisions.
- [Run behavior outside HTTP](/execution) from workers, jobs, tasks, scripts,
or application tools.
- [Build with AI](/ai) when the application needs coding-agent workflows,
machine-facing tools, MCP, or evaluations.
You can ignore those guides until the corresponding need appears. They extend
the same contracts, use cases, ports, and explicit wiring rather than adding a
second architecture.
Minor releases can still change Tenchi's public API. Read [stability and
releases](/stability) before adopting it for a long-lived service.
---
# Build your first Tenchi app
Source: https://tenchi.io/getting-started
Create a working application, call its API, follow one request through the
code, and change its behavior in a plain async function.
## Requirements
Tenchi requires Python 3.12 or newer. The commands in this guide use
[uv](https://docs.astral.sh/uv/) to create the environment and run the
application.
## Create the application
```shell
uvx tenchi new my_app
cd my_app
uv sync
```
The generated application already contains a todos feature and SQLite
persistence. Start it:
```shell
uv run tenchi dev
```
## Call the API
From another terminal, create and list todos:
```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 operation returns status `201`, a validated `Todo` body, and a
`Location` header. The list operation returns the saved todo. Restart the
server and list again to confirm that SQLite persisted it.
Open [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs) to inspect and call
the same API through Swagger UI.
## Follow the request through four files
The create operation has four application-facing parts:
| Path | Responsibility |
| --- | --- |
| `app/features/todos/schemas.py` | Defines the validated request and response data |
| `app/features/todos/contracts.py` | Declares `POST /todos` and its HTTP behavior |
| `app/features/todos/use_cases/create_todo.py` | Implements the application behavior |
| `app/features/todos/routes.py` | Binds the contract to the use case |
The contract and route handle the boundary. The use case stays ordinary
Python:
```python
async def create_todo(request: CreateTodo, context: AppContext) -> Todo:
return await context.todos.create(title=request.title)
```
`context.todos` is a `TodoRepository` port owned by the application. The
running server supplies a SQLite adapter; the direct use-case test supplies a
memory adapter. The use case does not need to know which one it received.
The generated project also includes server wiring, the SQLite and memory
adapters, and an HTTP test. You do not need to understand or edit them to
work on this operation.
## Change the behavior
Trim surrounding whitespace before saving a title. In
`app/features/todos/use_cases/create_todo.py`, change the repository call:
```diff
async def create_todo(request: CreateTodo, context: AppContext) -> Todo:
- return await context.todos.create(title=request.title)
+ return await context.todos.create(title=request.title.strip())
```
Update the direct test in
`app/features/todos/tests/test_create_todo.py` so it proves the new behavior:
```diff
- todo = await create_todo(CreateTodo(title="Buy milk"), context)
+ todo = await create_todo(CreateTodo(title=" Buy milk "), context)
assert todo.title == "Buy milk"
```
Run that focused test:
```shell
uv run pytest app/features/todos/tests/test_create_todo.py
```
The test calls the use case directly with a memory repository. It does not
start an HTTP server or open SQLite.
## Check the application
```shell
uv run tenchi check
```
`tenchi check` runs the generated project's formatting, linting, type,
behavior, architecture, and boundary checks. The command reports every failed
step so one run gives you the complete repair list.
Your first change touched one plain function and one focused test. The contract,
route, persistence wiring, OpenAPI, and HTTP behavior remained aligned without
requiring changes.
## Continue from the core
- [How Tenchi works](/concepts) explains the small application model.
- [Build a feature end to end](/build-a-feature) adds a persisted operation
across a port, two adapters, a route, tests, and the OpenAPI snapshot.
- [Prepare for production](/production) covers operational decisions when the
application is ready to ship.
---
# Build a feature end to end
Source: https://tenchi.io/build-a-feature
Add a `PATCH /todos/{todo_id}/complete` operation to the generated application.
The change crosses every layer involved in persisted behavior: the HTTP
contract, the repository port, the memory and SQLite adapters, the use case,
the route, a direct test, an HTTP test, and the OpenAPI snapshot.
Complete [Build your first Tenchi app](/getting-started) first, and run every
command in this guide from the application root.
## Declare the input and the HTTP operation
Add the path model to `app/features/todos/schemas.py`:
```python
class CompleteTodoParams(BaseModel):
todo_id: str
```
Replace the imports in `app/features/todos/contracts.py` with:
```python
from pydantic import BaseModel, Field
from tenchi.contracts import contract
from app.shared.errors import todo_not_found
from .schemas import CompleteTodoParams, CreateTodo, Todo
```
Add the operation beneath the existing contracts:
```python
complete_todo_contract = contract(
method="PATCH",
path="/todos/{todo_id}/complete",
params=CompleteTodoParams,
response=Todo,
errors=(todo_not_found,),
summary="Complete a todo",
tags=("todos",),
)
```
The contract says that the operation accepts one validated path parameter,
returns a `Todo`, and may expose the existing `TODO_NOT_FOUND` application
error. It does not yet implement or register the operation.
## Extend the repository port
Add the new capability to `TodoRepository` in
`app/features/todos/ports.py`:
```python
class TodoRepository(Protocol):
async def create(self, *, title: str) -> Todo: ...
async def list(self) -> list[Todo]: ...
async def complete(self, todo_id: str) -> Todo | None: ...
```
Returning `None` keeps storage concerns out of the public error model. The use
case will translate that result into the declared application error.
## Implement both adapters
Add this method to `MemoryTodoRepository` in
`app/infra/memory_todo_repository.py`:
```python
async def complete(self, todo_id: str) -> Todo | None:
todo = self._todos.get(todo_id)
if todo is None:
return None
completed = todo.model_copy(update={"completed": True})
self._todos[todo_id] = completed
return completed
```
The memory adapter gives direct tests a deterministic implementation without a
database.
Add the corresponding method to `SqliteTodoRepository` in
`app/infra/sqlite_todo_repository.py`:
```python
async def complete(self, todo_id: str) -> Todo | None:
await self._connection.execute(
"UPDATE todos SET completed = 1 WHERE id = ?",
(todo_id,),
)
cursor = await self._connection.execute(
"SELECT id, title, completed FROM todos WHERE id = ?",
(todo_id,),
)
row = await cursor.fetchone()
return _row_to_todo(row) if row is not None else None
```
The request-scoped context commits this update only after the use case and
response validation succeed. A failure rolls the transaction back.
## Implement the use case
Create `app/features/todos/use_cases/complete_todo.py`:
```python
from tenchi.errors import AppError
from app.server.context import AppContext
from app.shared.errors import todo_not_found
from ..schemas import CompleteTodoParams, Todo
async def complete_todo(
params: CompleteTodoParams,
context: AppContext,
) -> Todo:
todo = await context.todos.complete(params.todo_id)
if todo is None:
raise AppError(todo_not_found, details={"todo_id": params.todo_id})
return todo
```
The use case knows the application error but not its HTTP status or envelope.
The same function can later run from a task, a tool, or a script without
changes.
`uv run tenchi make use-case todos complete_todo --from-contract
app.features.todos.contracts:complete_todo_contract --dry-run` previews a
use case and a failing test derived from the contract. Add `--plan` when a
coding agent should prove the generated change was completed; see [verify a
generated change](/change-plans).
The generated starter has no authenticated subject in `AppContext`, so every
caller can complete every todo. Do not infer ownership from `todo_id`. When
the application gains users or tenants, enrich the context in an
authentication hook and add owner-scoped repository methods. The
[authentication guide](/authentication) shows that flow.
## Test the use case directly
Create `app/features/todos/tests/test_complete_todo.py`:
```python
from app.features.todos.schemas import CompleteTodoParams, CreateTodo
from app.features.todos.use_cases.complete_todo import complete_todo
from app.features.todos.use_cases.create_todo import create_todo
from app.infra.memory_todo_repository import MemoryTodoRepository
from app.server.context import AppContext
async def test_complete_todo() -> None:
context = AppContext(todos=MemoryTodoRepository())
created = await create_todo(CreateTodo(title="Buy milk"), context)
completed = await complete_todo(
CompleteTodoParams(todo_id=created.id),
context,
)
assert completed.id == created.id
assert completed.completed is True
```
Run it:
```shell
uv run pytest app/features/todos/tests/test_complete_todo.py
```
The test exercises the use case through the feature-owned port with the memory
adapter. It does not start a server or open SQLite.
## Bind the route
In `app/features/todos/routes.py`, add `complete_todo_contract` to the contract
imports and `complete_todo` to the use-case imports:
```python
from .contracts import (
CreatedTodoHeaders,
complete_todo_contract,
create_todo_contract,
list_todos_contract,
)
from .use_cases.complete_todo import complete_todo
```
Then add the binding to the existing `route_group()`:
```python
routes = route_group(
route(
create_todo_contract,
create_todo,
response_headers=create_todo_headers,
),
route(list_todos_contract, list_todos),
route(complete_todo_contract, complete_todo),
)
```
Importing the module now proves that the `params`, `context`, and return
annotations match the contract.
## Exercise the HTTP boundary
In `tests/test_http.py`, add the new contract and parameter model to the
existing imports:
```python
from app.features.todos.contracts import (
complete_todo_contract,
create_todo_contract,
)
from app.features.todos.schemas import CompleteTodoParams, CreateTodo
```
Then add an integration test that uses the SQLite-backed application:
```python
async def test_complete_todo_persists(tmp_path: Path) -> None:
database_path = str(tmp_path / "todos.db")
async with open_client(build_app(database_path)) as client:
created = await client.call(
create_todo_contract,
request=CreateTodo(title="Buy milk"),
)
completed = await client.call(
complete_todo_contract,
params=CompleteTodoParams(todo_id=created.id),
)
async with open_http(build_app(database_path)) as http:
listed = await http.get("/todos")
assert completed.id == created.id
assert completed.completed is True
assert listed.json() == [completed.model_dump()]
```
This test crosses route dispatch, request validation, the SQLite adapter,
transaction commit, result validation, and HTTP serialization. The direct test
remains the faster place for behavior and failure cases.
## See what the contract bought you
The contract is also the source of the published OpenAPI document, so Tenchi
can tell you what this change means for existing callers. Compare the composed
API with the committed snapshot:
```shell
uv run tenchi openapi --diff openapi.json
```
The report lists one additive change: a new operation that no current caller
depends on. Had you renamed a field or removed a response, the same command
would report a breaking change before anything shipped. Accept the additive
change by writing the new snapshot, then run the complete check:
```shell
uv run tenchi openapi --write openapi.json
uv run tenchi check
```
Run `uv run tenchi dev`, create a todo, and call
`PATCH /todos/{todo_id}/complete` to see the result yourself.
Continue with [authentication and authorization](/authentication) when the
operation must act for a verified user, or [databases and transactions](/database)
when adapting this pattern to another datastore.
---
# How Tenchi works
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. Five pieces are
enough to serve an HTTP API. Learn them once; everything else in Tenchi is
another entrypoint to the same pieces.
## Contract
A frozen declaration of one HTTP operation: method, path, validated inputs,
successful responses, and the application errors callers may observe. A
contract contains no handler logic.
## Route
A binding between a contract and a use case. `route()` inspects the use-case
signature immediately, so a mismatched parameter name or annotation fails 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 never receives a Starlette request, so the same function runs
unchanged from HTTP, tests, scripts, and workers.
## Port and adapter
A port is an interface the application needs, declared with `typing.Protocol`.
The feature owns the port; infrastructure supplies an adapter that implements
it. A SQL repository and a memory test double are both adapters for the same
port, so the application stays independent of any 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 at the server
composition root. Tenchi has no service locator or dependency-injection
container.
## 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.
## Optional extensions
None of the following is required to define and serve an HTTP API. Each one
gives an existing use case another entrypoint or another rule, and each has
its own guide.
- A **hook** runs at the HTTP boundary before input validation. It can
authenticate a request and return an enriched context. See
[authentication](/authentication).
- A **policy** is a pure function that answers an authorization question
inside a use case. See [authentication](/authentication).
- A **task** gives a use case a stable name for backfills, repairs, and
maintenance commands. See [operational tasks](/tasks).
- A **job** is a durable, validated message bound to a consumer use case. See
[background jobs](/jobs).
- A **tool** exposes a use case to machine callers with a stable name, typed
input and output, and safety annotations. See [application tools](/tools).
- An **evaluation** is a typed, thresholded suite that gates AI-powered
behavior. See [AI evaluations](/evaluations).
## 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.
---
# Add Tenchi 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
routes.py
runtime.py
```
These are the only modules Tenchi requires. Background jobs, operational tasks,
application tools, and evaluations each add one module under `app/server/`
when the application needs them; until then `tenchi map`, `tenchi check`, and
`tenchi verify` treat those boundaries as not configured. Preflight checks add
`app/server/preflight.py` when you adopt that deployment gate.
## 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).
## 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.
## Declare the verification policy
Create `tenchi.toml` at the application root:
```toml
schema_version = 1
[verify]
check = true
architecture = true
openapi = true
```
This makes the repository's definition of done explicit for humans, agents,
and CI. Stages you omit, such as `jobs`, `tools`, and `evaluations`, are
recorded as not configured until you add them. A project without this file
receives the same behavior from Tenchi's built-in policy, which requires those
three stages plus any optional boundary whose composition module exists, so
adding the file is a metadata-only adoption. Later verification compares it
with the selected Git baseline, retains any stronger historical requirement for
the current run, and rejects a removed or weakened policy.
## Create the baseline and validate the app
Before running verification, make expected local artifacts invisible to Git.
Tenchi's source digest deliberately includes every nonignored untracked path,
so a cache or development database created during a check otherwise invalidates
the [receipt](/change-plans#verification-terms). Keep source, snapshots,
lockfiles, and configuration visible; add only reproducible or
environment-local artifacts to `.gitignore`. For example:
```gitignore
__pycache__/
.venv/
.pytest_cache/
.ruff_cache/
.coverage
.coverage.*
*.log
*.db
*.db-shm
*.db-wal
```
Write the first canonical OpenAPI snapshot, then run the local project gates:
```shell
uv run tenchi openapi --write openapi.json
uv run tenchi check
uv run tenchi map
```
The snapshot establishes the application's compatibility baseline. Commit it
before using `tenchi verify`: a ref that predates `openapi.json` cannot support
a historical compatibility claim, so `verify` correctly fails when a required
baseline is absent. On the next change, compare with the committed baseline:
```shell
uv run tenchi verify --base-ref origin/main
```
Use `tenchi check` as the CI gate on the adoption change. After the baseline
commit reaches the target branch, run `tenchi verify` with the pull-request
base ref for later changes so CI checks both exact snapshot drift and
historical compatibility.
When you later add a durable job, application tool, or evaluation, add its
module, write its snapshot, and set the matching `[verify]` stage to `true` in
the same change. `tenchi verify` treats the missing historical snapshot as a
first adoption when the module did not exist at the baseline. [Background
jobs](/jobs), [application tools](/tools), and [AI evaluations](/evaluations)
each describe that step.
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.
---
# Structure your application
Source: https://tenchi.io/architecture
Use this structure when deciding where new code belongs and which direction its
dependencies may point. Tenchi organizes application behavior by feature while
keeping infrastructure and server composition explicit.
```text
app/
features//
contracts.py
schemas.py
ports.py
policy.py
routes.py
jobs.py
tasks.py
tools.py
evaluations.py
use_cases/
tests/
shared/
infra/
server/
context.py
hooks.py
webhooks.py
routes.py
jobs.py
runtime.py
preflight.py
evaluations.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. `evaluations.py` declares typed cases, metrics, and
provider-neutral evaluators for behavior the feature needs to gate.
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.
- `evaluations.py` composes evaluations with application lifecycle and context
wiring.
- `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.
Only `context.py`, `routes.py`, and `asgi.py` are required. The other server
modules exist when the application uses that capability. `tenchi map`,
`tenchi check`, and `tenchi verify` treat an absent default module for jobs,
tasks, tools, or evaluations as not configured; `tenchi preflight` requires
its module when you run it.
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
server composition -> evaluation declarations + application 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.
- Evaluation modules may import schemas, ports, policies, and use cases, but
never infrastructure or server composition. They describe cases and scoring;
provider adapters arrive through the runner's context.
- 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. Doctor's structural rules are
listed under [Constraints](#constraints).
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)
```
## Constraints
`tenchi doctor` enforces the dependency direction above and these structural
rules:
- `app/server/asgi.py`, `app/server/context.py`, and `app/server/routes.py`
must exist. The modules that compose jobs, tasks, tools, evaluations, and
preflight checks are optional.
- Imports inside feature package initializers are checked like any other
feature module.
- Modules directly under `app/` outside `features`, `shared`, `infra`, and
`server` are rejected.
- Symlinked application paths are rejected so results do not vary with
Python's directory-walking behavior.
- Test modules are exempt from the import rules.
---
# Declare an HTTP contract
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,
request_examples={"create": CreateTodo(title="Buy milk")},
response_examples={
"created": Todo(id="todo_123", title="Buy milk", completed=False)
},
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.
Every path, query, and header field needs one unambiguous wire encoding. Tenchi
rejects shapes that lack one when the route, server, or OpenAPI document is
composed. The exact rules are listed under [Constraints](#constraints).
## 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. Incompatible body annotations fail when the contract is declared;
for example, an object model cannot be paired with `text/plain`, and raw bytes
cannot be a JSON response body.
JSON response bodies are checked against their published schema and read back
through the declared type before the request scope commits. See [successful
responses](/responses) for that behavior and its rollback rules.
## Paginate collections
`tenchi.pagination` provides one small limit-and-offset vocabulary that stays
typed across the contract, use case, client, and OpenAPI document:
```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 with `items`, `total`, `limit`, and `offset`.
The use case builds the page from the values its port returns:
```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, so the
repository result and the public envelope stay aligned. Memory adapters may
slice a list, but production repositories should apply limit, offset, and
filters in the database or remote service and return the page plus the total
count through the port. The typed client returns `Page[Todo]`, and OpenAPI
includes the query constraints and item schema without another declaration.
## Give callers concrete examples
Use `request_examples=` and `response_examples=` when a caller would benefit
from seeing a complete valid exchange. Each mapping key is a stable,
human-readable example name. Values are ordinary instances of the declared
types—not hand-written JSON.
When OpenAPI is generated, Tenchi validates an isolated copy of each value,
serializes it with Pydantic and its wire aliases, revalidates that wire payload,
and verifies it against the schema it publishes. A stale value, a non-round-trip
validator/serializer pair, or a serializer that contradicts the schema fails
generation without including the example payload in the error. Examples appear
under the operation's request or response media type, where OpenAPI clients and
agents can discover them.
For a contract with `responses=`, put `examples={...}` on each
[`response()` definition](/responses) so every example stays attached to its
status and media type. Examples are public API documentation: use realistic
placeholder values, never credentials, personal data, or production payloads.
## 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).
`idempotency_key=True` makes an unsafe operation's retry guarantee explicit.
The contract must declare one required, non-empty string `Idempotency-Key`
header. Tenchi checks both Pydantic validation and serialization schemas and
publishes the guarantee in OpenAPI. The flag does not implement replay or
storage; follow [Make operations safe to retry](/idempotency) for that work.
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 HTTP work, waits for request-scope
cleanup, and returns a framework-owned `504`. It does not follow the use case
into tools, jobs, tasks, or direct execution.
- `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).
## Constraints
Tenchi enforces these rules when the contract is declared or when the
application, OpenAPI document, or typed client is composed.
- Path fields serialize as one non-null scalar value and cannot be nullable.
- Query fields serialize as one scalar value or an array of non-null scalar
values; arrays use repeated query keys. A field cannot combine scalar and
array alternatives because a single query value would be ambiguous.
Fixed-length tuples of scalar values remain supported.
- Request-header fields serialize as one scalar value. Their aliases must
normalize to unique, valid HTTP header names and cannot name transport-owned
headers such as `Content-Type`, `Content-Length`, or `Host`; body media types
belong to `request_media_type=`. Header values must be safe ASCII without
control characters or edge whitespace.
- Nested path or query objects, multi-valued headers, null-only fields, and
open-ended parameter mappings are rejected because they have no implicit
wire encoding.
- Nullable query and header fields must be omittable, usually through a `None`
default. A required nullable field cannot distinguish `None` from a missing
parameter and is rejected.
- Singular successful statuses must be between 200 and 399. Singular `204`,
`205`, and `304` outcomes must omit `response=` because those statuses cannot
carry a body.
- Text media types validate a string; other non-JSON media types validate
bytes or strings. An object model cannot be paired with `text/plain`, and raw
bytes cannot be a JSON response body. Charset-qualified text types are encoded
and decoded strictly, and an unsupported declared charset fails when the
contract is built.
- JSON response bodies must satisfy their published serialization schema and be
readable by the declared type. Response field aliases must be readable by the
same model: use `Field(alias=...)` or accept the serialization name in
`validation_alias`. See [response constraints](/responses#constraints).
- Before sending a typed request, `Client` passes the concrete path, query, and
header values through their HTTP encoding and the declared validator again,
including configured httpx defaults, so an inherited parameter cannot
silently replace an omitted value. A custom serializer must preserve both the
scalar-or-array shape and the validated value. An empty repeated value is
usable only when omission reconstructs the same default; otherwise the client
fails before I/O without including the parameter payload in the error.
---
# Write 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.
---
# Bind routes and compose the 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. Route templates are checked
during composition and matched by shape at runtime; the rules are listed under
[Constraints](#constraints).
## 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.
## Constraints
- `create_app()` 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.
- Matching is independent of declaration order. Literal and constrained paths
take precedence over broader parameterized paths.
- A `GET` contract handles `HEAD` automatically unless an explicit `HEAD`
contract matches.
- Contract paths are exact. If `/todos` is declared, `/todos/` reaches the
structured framework 404 response instead of redirecting. Declare both
spellings only when the API intends to support both.
- Observer failures are logged and cannot change the response or the use-case
result.
---
# Model successful 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.
For ordinary JSON responses, Tenchi serializes the value, checks those exact
bytes against the published serialization schema, and reads them back through
Pydantic before the request scope commits. A failure becomes a framework-owned
500 and rolls back scoped writes. The rules that make a response readable are
listed under [Constraints](#constraints).
## 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.
Presenters and header projectors may raise a declared `AppError`. It uses the
same application-error response as an error raised by the use case; undeclared
errors remain framework-owned 500 responses.
## 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",
examples={
"created": Todo(id="todo_123", title="Buy milk", completed=False)
},
)
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`. Examples belong on the individual definition so
OpenAPI attaches each one to the correct status and media type. Tenchi validates
and serializes them the same way as singular `response_examples=` values.
## 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)`.
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.
## Constraints
- Serialization is checked on the value actually returned, so an already-created
model whose fields or nested collections were mutated after construction is
checked too. Custom serializers must produce schema-valid JSON that the
declared type can read back.
- Response field aliases must be readable by the same model. Prefer
`Field(alias="wire_name")` for a shared input and output name. With separate
validation and serialization aliases, include the output name in the accepted
validation aliases, for example `AliasChoices("input_name", "wire_name")`.
Unreadable aliases fail during application composition, OpenAPI generation,
and client preflight, including inside nested response types.
- JSON Schema validation resolves local references only and never fetches
remote schemas during a request. Use self-contained schemas for custom
response types.
- A required response-header field cannot accept `None`, because a null value
has no HTTP header representation. Make the field optional when omitting the
header is valid.
- An empty passthrough definition must present a Starlette response with a
materialized empty body.
- Presenters and header projectors may raise a declared `AppError`; undeclared
errors remain framework-owned 500 responses.
---
# Expose honest application 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.errors import AppError, ErrorDef
write_temporarily_unavailable = ErrorDef(
code="WRITE_TEMPORARILY_UNAVAILABLE",
status=503,
message="Todo writes are temporarily unavailable",
headers=("Retry-After",),
)
create_todo_contract = contract(
method="POST",
path="/todos",
request=CreateTodo,
response=Todo,
errors=(write_temporarily_unavailable,),
)
async def create_todo(request: CreateTodo, context: AppContext) -> Todo:
if not await context.todos.accepting_writes():
raise AppError(
write_temporarily_unavailable,
headers={"Retry-After": "30"},
)
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. The built-in `RATE_LIMITED`
error uses the same mechanism for its `Retry-After` value. 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.
OpenAPI narrows `code` to the exact values available at each response status
and documents the error-source header. Every operation includes the
framework-owned `500` response used for unexpected or undeclared failures.
Authentication errors appear when you declare them on the protected contract
or route group; configuring an OpenAPI security scheme does not invent an
application-specific 401 or 403.
## Client symmetry
The typed client raises declared `AppError` values. An undocumented status,
invalid envelope, wrong error source, mismatched media type, or invalid success
body or headers 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.
---
# Authenticate and authorize requests
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.
---
# Test at the right boundary
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 boundary snapshots 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).
When the application adds durable jobs, application tools, or evaluations,
add the matching test so their snapshots stay exact too:
```python
from tenchi.cli import main
def test_tool_snapshot_is_current() -> None:
assert main(["tools", "--check", "tools.json"]) == 0
```
The job and evaluation variants call `["jobs", "--check", "jobs.json"]` and
`["eval", "snapshot", "--check", "evaluations.json"]`. `tenchi new --full`
generates all three. Run the matching `--diff` command before replacing any of
these snapshots: the exact test proves reproducibility, and the historical
diff proves compatibility. See [background jobs](/jobs), [application
tools](/tools), and [AI evaluations](/evaluations) for each workflow.
---
# Call contracts with the 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=`.
Path values and response aliases are checked before any request is sent; see
[Constraints](#constraints).
## 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",),
retry_on_statuses=(502, 503, 504),
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. Prefer selecting
declared application errors by stable code in `retry_on`. A status in
`retry_on_statuses` retries every response with that status—including a
declared application error—so add one only when an intermediary may return the
same transient failure without Tenchi's error envelope.
Tenchi uses exponential backoff with bounded jitter. A `Retry-After` value from
a selected status or declared error can increase the wait, but never beyond
`max_delay_seconds`. The total timeout covers attempts and backoff; exceeding it
raises `RetryTimeoutError`. Caller cancellation always remains cancellation.
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.
Any response that violates this boundary raises `UnexpectedResponseError`.
The exception includes the contract name, HTTP status, and a stable reason, but
never retains the response body or headers. This includes malformed JSON and
declared success bodies or headers that fail Pydantic validation, so logs and
agent output do not accidentally retain dependency payloads.
## 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.
## Constraints
- Path values must match the declared Starlette converter before any request is
sent. Ordinary `{name}` or `{name:str}` parameters cannot contain `/`; use an
explicit `{name:path}` converter for a nested path. `.` and `..` segments are
rejected even inside path converters. Percent encoding does not make a slash
safe for a single-segment parameter, because the server routes the decoded
path.
- Response field aliases must be accepted by the response model's validation
aliases. A model with only `serialization_alias="wire_name"` cannot read its
own output unless its validation configuration also accepts that name. Tenchi
rejects these declarations before I/O. See [response
constraints](/responses#constraints).
- `retry_on_statuses` accepts unique values from 400 through 599. Every other
unexpected status remains terminal.
- A `Retry-After` value must be a decimal-integer delay or an HTTP date.
Malformed and past values are ignored, and no value can extend the wait
beyond `max_delay_seconds`.
- Invalid success bodies or headers are never retried. Invalid media types or
error envelopes are retried only when their HTTP status is explicitly
selected. Contract drift and bad data stay terminal; deliberate recovery from
raw proxy and load-balancer failures remains possible.
- Retrying POST, PUT, PATCH, or DELETE requires `allow_unsafe_methods=True`.
---
# Prepare the application for production
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 guide turns that model into an ordered production checklist 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 |
| Idempotency | Canonical fingerprints, durable store transitions, typed replay, and declared errors | Implement storage in the command transaction and choose key scope and retention |
| Outbound retries | Opt-in attempt limits, declared-error and raw-status selection, bounded backoff, `Retry-After`, total deadlines, cancellation, and payload-safe attempt outcomes | Mark transient errors and intermediary statuses, choose timing, and authorize unsafe methods only when the operation is retry-safe |
| 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 |
| AI quality | Typed cases, normalized thresholds, bounded execution, redacted reports, and optional token/cost budgets | Choose providers, prompts, datasets, scorers, judge models, and acceptable variance |
| 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. [Gate model-backed behavior with explicit evaluations](/evaluations).
10. [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;
- an explicitly budgeted `tenchi eval run` gate for model-backed behavior,
separate from deterministic source checks;
- `tenchi verify --base-ref ` for checks, strict architecture
evidence, and OpenAPI, job-message, application-tool, and evaluation-policy comparisons
against one immutable release baseline.
The [deployment guide](/deployment) turns this list into the final release
sequence.
---
# Load 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.
This page covers values the running application consumes. The repository's
`tenchi.toml` does not hold runtime settings or secrets; it declares which
evidence [`tenchi verify`](/cli#verify-a-completed-change) must produce before
a change is complete.
## 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.
---
# Own database 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.
---
# Make operations retry-safe
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,
idempotency_key=True,
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`.
`idempotency_key=True` is a checked contract promise, published in OpenAPI as
`x-tenchi-idempotency-key`, so generated clients and agents can tell that
retries with the same key describe one logical command. The flag documents the
guarantee; the use case and durable store below implement it. The header and
compatibility rules are listed under [Constraints](#constraints).
## 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.
Before completing a reservation, `run_idempotently()` serializes an isolated
copy of the result with Pydantic's round-trip mode and revalidates the stored
bytes. The reconstructed value must equal the original typed result, so
`Json[T]` fields can replay while lossy serializers and omitted required fields
raise `IdempotencyResultError` before completion. The reservation is abandoned
on failure; keep the store and application writes in the same transaction so
those writes roll back too.
`fingerprint()` validates against `annotation=` before hashing canonical JSON.
Equivalent validated values therefore share a fingerprint even when mapping
order or coercible input spelling differs. Pass a concrete annotation when
Pydantic owns a non-JSON representation such as a datetime.
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.
A durable adapter must share the command's transaction when the idempotency
record protects database writes. That lets concurrent requests either commit
the work once or replay its committed result. Run
[`verify_idempotency_store()`](/testing#verify-store-adapters) against the
adapter before relying on it, then separately test rollback with the
application writes it protects.
## Constraints
- `idempotency_key=True` is allowed only on unsafe methods and requires exactly
one required, non-empty string header whose wire name is `Idempotency-Key`.
Tenchi checks both the Pydantic validation and serialization schemas.
- Removing the promise is a breaking OpenAPI compatibility change. Adding it is
additive only when the required header already exists; adding that required
header to an established operation remains breaking for existing callers.
- `IDEMPOTENCY_CONFLICT` and `IDEMPOTENCY_IN_PROGRESS` follow the [honesty
rule](/errors#the-honesty-rule). Undeclared, they become framework-owned 500
responses instead of undocumented 409 responses.
- A fingerprinted value must validate back to the same untouched Python value
and serialize to the same canonical JSON again. Lossy or mutating
serializers, dynamic or undeclared serializer output, and `Any` values such
as non-finite floats or bytes are rejected so they cannot collide with `null`
or strings or give one request changing fingerprints.
- Without `annotation=`, every non-JSON root type, including Pydantic models
and string enums, includes its runtime type in the fingerprint.
- `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.
- A completed result must reconstruct the same typed value from its stored
bytes. Lossy serializers and omitted required fields raise
`IdempotencyResultError` before completion and abandon the reservation.
---
# Limit application operations
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 and returns either:
- `RateLimitPermit(limit, remaining, reset_after_seconds)` for accepted work;
- `RateLimitExceeded(limit, retry_after_seconds)` for rejected work.
The window semantics an adapter must implement are listed under
[Constraints](#constraints).
## 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,
)
```
Tenchi validates the policy values before calling the store; see
[Constraints](#constraints).
## 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.
## Constraints
- `limit` and `cost` must be positive integers, and `cost` cannot exceed
`limit`. `window_seconds` must be finite and greater than zero.
- 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.
- `enforce_rate_limit()` validates the store's returned values. A malformed
adapter response raises `RateLimitStoreError` rather than producing a
misleading allowance or `Retry-After` value.
- `MemoryRateLimitStore` is concurrency-safe within one process only and loses
every window on restart.
---
# Receive 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.
---
# Design for retries and partial failure
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, `JobPayloadError`, 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.
---
# Run use cases outside HTTP
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.
Invalid input raises `ExecutionInputError` before the context opens. Its
`issues` contain stable validation kinds without rejected values, field paths,
or rendered validation messages, so a worker can classify and log it without
persisting application payloads.
`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`.
## Set an entrypoint-neutral deadline
An HTTP contract's `timeout=` applies only while that contract runs through a
route. It does not limit the same use case when a tool, job, task, script, or
direct caller invokes it. Put a deadline around latency-sensitive application
work when every entrypoint needs the same bound:
```python
import asyncio
from tenchi.errors import AppError
async def answer_question(
request: AnswerQuestion,
context: AppContext,
) -> Answer:
try:
async with asyncio.timeout(25):
return await context.answers.generate(question=request.question)
except TimeoutError as exc:
raise AppError(answer_provider_unavailable) from exc
```
Declare `answer_provider_unavailable` on every HTTP contract or application
tool that may expose it. Configure the provider SDK's own connection and
request timeout below the application deadline so it can close network
resources before the application maps the failure. A caller may still impose a
shorter outer deadline, and cancellation must continue to propagate through the
use case and its cleanup scopes.
## 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.
---
# Dispatch validated 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 bytes satisfy the schema published in `jobs.json` and can be read back
strictly by the declared consumer type. 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: ...
```
If a custom serializer emits a different wire shape, `job_message()` raises
`JobBindingError` before enqueueing it. Keep custom serialization aligned with
the JSON Schema generated from the request annotation.
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.
## Protect stored messages with a snapshot
Write the canonical manifest after registering a job:
```shell
uv run tenchi jobs --write jobs.json
```
The manifest contains stable names, descriptions, and input JSON Schemas. It
never contains queued payloads or handler results. Commit `jobs.json`, then
check exact drift locally and in CI (`tenchi check` runs this step once
`app/server/jobs.py` exists, and `jobs = true` under `[verify]` in
`tenchi.toml`, declared in the same change, makes `tenchi verify` enforce the
historical comparison):
```shell
uv run tenchi jobs --check jobs.json
```
Before accepting a changed snapshot, compare it with the current file or a Git
baseline:
```shell
uv run tenchi jobs --diff jobs.json
uv run tenchi jobs --diff-ref origin/main --snapshot jobs.json
```
When adopting the manifest for the first time and the selected Git ref truly
predates `jobs.json`, add `--allow-missing-baseline` to that one `--diff-ref`
comparison. The report records the missing baseline as metadata. Future
comparisons fail if the historical file is absent. `tenchi verify` needs no
override when `app/server/jobs.py` itself did not exist at the baseline.
Removing a job or narrowing the payloads its consumer accepts is breaking.
Adding a job or widening accepted payloads is additive; description-only
changes are metadata. Changes the analyzer cannot prove safe require review.
If a breaking payload change is intentional, declare a new job name so workers
can continue consuming messages stored under the old contract during rollout.
Compatibility is directional: it proves that the new consumer accepts messages
valid under the historical manifest. Deploy that consumer before producers can
enqueue the new shape. If rollback must restore an older consumer after new
messages exist, keep the emitted shape acceptable to both versions or use a new
job name.
## 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.
`JobDispatcher` does not choose an execution timeout. The worker should apply a
bounded delivery deadline, while a use case that needs the same limit in every
entrypoint should enforce an [entrypoint-neutral application
deadline](/execution#set-an-entrypoint-neutral-deadline). Cancellation must
still reach the dispatcher so the context can roll back before the worker
releases or retries the claim.
| Failure | Typical worker decision |
| --- | --- |
| `JobNotFoundError` | Dead-letter; deploying the correct consumer may make a later replay possible |
| `JobPayloadError` | Dead-letter; its payload-safe issues identify why the stored message 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.
Continue with [Retries and background work](/reliability) to connect typed
producer messages and registered dispatchers to a transactional outbox,
rollback rules, dead-lettering, and worker-owned retries.
---
# Run validated 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. Direct `TaskRunner.run()` callers receive a
payload-safe `TaskInputError` for invalid input; its `issues` expose only stable
validation kinds.
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.
`TaskRunner` does not impose an execution timeout. Let the operator environment
set an outer process deadline, and put an [entrypoint-neutral application
deadline](/execution#set-an-entrypoint-neutral-deadline) inside behavior that
must remain bounded when called through HTTP, tools, jobs, and tasks alike.
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.
---
# Observe behavior without exposing payloads
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, job-message, application-tool, or evaluation-policy 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.
---
# Verify a deployment environment
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
Create `app/server/preflight.py` exposing `checks`, the module `tenchi
preflight` loads by default (`tenchi new --full` generates an empty one). 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",
),
)
```
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.
Naming and shape rules are listed under [Constraints](#constraints).
## 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": 12,
"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.
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,
job-message, application-tool, and evaluation-policy 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.
## Constraints
- Check names use dotted `snake_case` and remain stable for deployment
automation. Failure codes use `SCREAMING_SNAKE_CASE`; when omitted, Tenchi
derives one such as `PREFLIGHT_DATABASE_SCHEMA_FAILED`.
- Each check is a zero-argument async function that returns `None` on success
and raises on failure. A check that returns a value is reported as failed and
the value is discarded.
- A check must propagate cancellation so its client and connection cleanup can
finish when it times out.
- `--timeout` caps every declared timeout and never extends one.
- Descriptions and failure codes come from the declaration and appear in
results, so keep them free of credentials and tenant data.
- Preflight receives no application context and does not start the
application lifespan.
---
# Deploy the ASGI application
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.
## Separate liveness from readiness
Use different routes for the two deployment decisions. Liveness answers
whether the process is running and responsive, so it must not depend on a
database, cache, model provider, or another network service. Readiness answers
whether this process should receive traffic and may check dependencies required
by the operations it serves:
```python
from tenchi.health import health_route
liveness = health_route(path="/live")
readiness = health_route(
path="/ready",
checks={"database": check_database},
check_timeout=2.0,
)
```
Register both routes at the composition root. Configure the platform's
liveness probe to call `/live` and its readiness probe to call `/ready`. A
database outage can then remove this process from traffic without causing a
restart loop.
Readiness checks run concurrently and return `503` when any check is unhealthy
or exceeds `check_timeout`. Synchronous checks run in worker threads so
blocking I/O cannot hold the event loop past the health deadline; the
underlying thread may finish after the timed-out response, so prefer async
clients with their own deadlines. An async check that suppresses cancellation
may also finish in the background, but it cannot delay the response. Keep
checks read-only and independently bounded. Health routes are 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. A contract's
`timeout=` bounds only its HTTP route. Put an [entrypoint-neutral
deadline](/execution#set-an-entrypoint-neutral-deadline) around application work
that needs the same bound through tools, jobs, tasks, and direct execution.
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 every snapshotted boundary with one
baseline commit from the version currently deployed:
```shell
uv run tenchi verify --base-ref "$BASE_SHA"
```
`verify` discovers the literal OpenAPI title, version, description, and
security configuration from `app.server.routes:api_routes`. It runs the same
deterministic checks, rejects an application map with diagnostics or unresolved
relationships, and compares
OpenAPI, durable job messages, application tools, and evaluation policy with
the resolved commit. It also compares `tenchi.toml` with that commit and keeps
the stronger current-or-deployed requirement active, so a release cannot avoid
a gate by weakening it in the same change. Retain the complete
[receipt](/change-plans#verification-terms) before making another edit: its source commit and tree digest identify the exact
application state that passed, and verification fails if that state changes
while the command is running.
Use its metadata flags only when the application does not keep those values as
literal route-module declarations.
Missing historical job or evaluation snapshots fail by default. If that commit
predates only `jobs.json`, add `--allow-missing-job-baseline` once and confirm
the result records a `job manifest baseline` metadata change. Use
`--allow-missing-evaluation-baseline` in the same way for `evaluations.json`.
Remove either override after its baseline lands.
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. For model-backed behavior, run the explicitly budgeted
[`tenchi eval run`](/evaluations) gate with the provider configuration this
release will use.
7. Start or update [background workers](/reliability) that can consume messages
from both application generations. Deploy compatible consumers before new
producers receive traffic.
8. Wait for startup and readiness checks, then shift traffic gradually.
9. Smoke-test a public route and an authenticated operation through the
production edge.
10. 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. AI evaluation execution
remains separate because it may be nondeterministic and incur provider cost;
the declared policy is still checked locally.
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.
---
# Review API compatibility
Source: https://tenchi.io/openapi
Tenchi derives OpenAPI 3.1 from composed contracts. The document describes the
same types, headers, errors, media types, examples, visibility, retry guarantees,
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.
## Discover error behavior
Every error response includes the stable Tenchi envelope and narrows `code` to
the exact values that status may return. It also documents the required
`x-tenchi-error-source` header as `app`, `framework`, or both. A generated
client or agent can therefore distinguish declared application outcomes from
framework failures without parsing response descriptions.
Every operation documents `500` with the framework-owned
`INTERNAL_SERVER_ERROR` code. Tenchi uses that response for unexpected
exceptions and undeclared application errors.
Security schemes describe how to authenticate; they cannot determine which
authentication errors an application uses. Declare authentication failures on
the protected route group so their statuses and codes appear on each affected
operation:
```python
private_routes = route_group(
project_routes,
task_routes,
errors=(unauthorized,),
)
```
## Teach clients and agents how to call an operation
Named `request_examples=` and `response_examples=` values flow from a contract
to the corresponding OpenAPI Media Type Object. Tenchi runs them through the
declared Pydantic validator and serializer, revalidates the serialized wire
payload through the receiving boundary, and checks the resulting JSON against
the published schema. This keeps aliases, custom validators, and custom
serializers honest while giving external clients and agents concrete payloads
they can adapt.
```python
create_todo_contract = contract(
method="POST",
path="/todos",
request=CreateTodo,
response=Todo,
status=201,
request_examples={"create": CreateTodo(title="Buy milk")},
response_examples={
"created": Todo(id="todo_123", title="Buy milk", completed=False)
},
)
```
Changing examples is metadata-only for compatibility. Do not put secrets,
personal data, or production payloads in them because the OpenAPI document is
often public.
Contracts marked `idempotency_key=True` or `webhook=True` publish Tenchi
extensions that describe caller-visible guarantees. Their compatibility rules
are listed under [Constraints](#constraints).
## 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 --write openapi.json
uv run tenchi openapi --check openapi.json
```
`--write` produces deterministic, key-sorted JSON. `--check` is an exact
equality check suitable for a repository test. By default the command loads
`app.server.routes:api_routes` and discovers literal `OPENAPI_TITLE`,
`OPENAPI_VERSION`, `OPENAPI_DESCRIPTION`, and `OPENAPI_SECURITY` declarations
from that module. Use the corresponding flags only to override this convention.
## Classify changes
```shell
uv run tenchi openapi --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.
## Try a breaking and additive change
In a generated application, open `app/features/todos/schemas.py` and tighten
the title constraint:
```diff
class CreateTodo(BaseModel):
- title: str = Field(min_length=1)
+ title: str = Field(min_length=3)
```
Compare the current contracts with the checked-in snapshot:
```shell
uv run tenchi openapi --diff openapi.json
```
The command fails because existing callers can send titles that the new
constraint rejects. Restore `min_length=1`, then add an optional field:
```python
class CreateTodo(BaseModel):
title: str = Field(min_length=1)
notes: str | None = None
```
The same diff command now reports an additive change. After reviewing it,
update the snapshot and check the application:
```shell
uv run tenchi openapi --write openapi.json
uv run tenchi check
```
Use `tenchi verify --base-ref ` when the repository also needs
one [receipt](/change-plans#verification-terms) for its checks, application
map, verification policy, and every versioned boundary.
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 --diff-ref "$BASE_SHA" --snapshot openapi.json
uv run tenchi openapi --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.
Programmatic consumers can import `analyze_openapi_compatibility()` from
`tenchi.compatibility` and inspect its structured `CompatibilityReport`.
## Constraints
- Keep the route module's literal `OPENAPI_*` declarations stable across
generation, comparison, and snapshot checks. Explicit CLI flags take
precedence when an application uses a nonstandard target or metadata source.
- An unsafe contract marked `idempotency_key=True` must declare a required,
non-empty string `Idempotency-Key` header. It then emits
`x-tenchi-idempotency-key: Idempotency-Key`. Adding the guarantee is additive
when the required header already exists; adding that header to an
established operation is breaking; removing the guarantee is breaking because
callers may rely on safe logical retries. The extension describes behavior
and does not replace the idempotency store or use case.
- Contracts marked `webhook=True` emit `x-tenchi-webhook: true`. Adding that
requirement is breaking because existing callers must begin signing
deliveries; removing it is additive.
- Changing `request_examples=` or `response_examples=` is metadata-only.
- Security schemes describe how to authenticate. They never add authentication
error responses; declare those on the protected contract or route group.
---
# Build with AI
Source: https://tenchi.io/ai
Tenchi supports two different roles for AI without creating two application
architectures:
| Goal | Tenchi's role |
| --- | --- |
| Let a coding agent change the backend | Give the agent deterministic inspection, previewable generation, structured checks, and historical verification |
| Add AI behavior to the backend | Expose ordinary use cases as typed tools, keep providers behind ports, and gate model behavior with evaluations |
Both paths lead back to the same plain async use cases and explicit dependency
wiring. Model-generated input never supplies identity or infrastructure, and a
coding agent does not need a hidden framework runtime to understand the
application.
## Let an agent change the backend
A coding agent works through the same development loop as a person:
```shell
uv run tenchi map --feature projects --json
uv run tenchi make use-case projects create_project \
--from-contract app.features.projects.contracts:create_project_contract \
--dry-run --json
uv run tenchi check --json
uv run tenchi verify --base-ref origin/main --json
```
The map explains what exists and how it is connected. The preview derives a
use-case boundary from the contract without writing files. `check` reports the
complete local repair list, and `verify` compares the finished application with
a Git baseline.
Generated applications include an `AGENTS.md` and project-local MCP
configuration. Use the [coding-agent workflow](/agents) with any agent that can
read files and run commands, or [connect an MCP-aware coding agent](/mcp) to the
same operations.
For contract-driven generation, a [change plan](/change-plans) can bind the
requested contract, generated files, route wiring, and one test target to the
final verification receipt. The [receipt](/change-plans#verification-terms) proves that the structure was
completed and its test ran. When the code producer is untrusted, acceptance
tests owned outside the repository remain the right check.
## Add AI behavior to the backend
Keep model calls behind an application-owned protocol, just like a database or
external API:
```python
from typing import Protocol
from .schemas import Answer, AnswerRequest
class AnswerGenerator(Protocol):
async def answer(self, request: AnswerRequest) -> Answer: ...
```
The use case owns authorization, retrieval, business rules, and the decision to
call that port. Infrastructure selects the provider and model at composition.
The same use case can serve HTTP and an AI-facing tool:
```python
from app.shared.errors import question_not_answerable
from tenchi.tools import tool, tool_group, tool_handler
from .schemas import Answer, AnswerRequest
from .use_cases.answer_question import answer_question
answer_tool = tool(
"knowledge.answer",
request=AnswerRequest,
result=Answer,
description="Answer from sources visible to the authenticated user.",
errors=(question_not_answerable,),
read_only=True,
open_world=False,
)
tools = tool_group(
tool_handler(answer_tool, answer_question),
)
```
The [application-tool boundary](/tools) validates input and output, publishes a
deterministic manifest, and applies the application's lifespan and context to
every call. It does not choose an agent SDK or model provider. Call tools
in-process from your preferred AI runtime or [serve them over authenticated
MCP](/tool-mcp).
## Keep production rules in the application
AI callers use the same production boundaries as every other caller:
- Authentication supplies identity through application-owned context wiring.
- Use cases and pure policies authorize every action.
- Tool safety annotations describe behavior but never grant permission.
- Idempotency, rate limits, deadlines, retries, jobs, and observability remain
ordinary application concerns.
- Declared errors stay stable; unexpected exceptions and undeclared application
errors remain behind a generic invocation failure.
This keeps model prompts and tool-selection logic from becoming an alternate
authorization or transaction layer.
## Gate behavior that deterministic tests cannot prove
When a provider or model can change behavior without a Python code change,
declare application-owned evaluation cases and metrics:
```shell
uv run tenchi eval list
uv run tenchi eval run
uv run tenchi eval snapshot --diff evaluations.json
```
[AI evaluations](/evaluations) run typed cases with bounded concurrency,
timeouts, thresholds, and optional token or cost budgets. The policy snapshot
contains no case inputs, so it can make a weakened gate visible in review
without running a model during every deterministic check.
Tenchi does not implement model turns, handoffs, prompt templates,
conversation memory, vector search, or RAG orchestration. Use the libraries
that fit your product behind application ports; use Tenchi to keep their
inputs, permissions, lifecycle, outcomes, and release gates explicit.
## See the complete shape
[Fieldnotes](/fieldnotes) combines owner-scoped ingestion, background indexing,
authenticated application tools, MCP, cited answers behind a provider port,
operational reindexing, preflight checks, and evaluation gates without model
credentials in CI.
Choose the next page from the role AI will play:
- [Let a coding agent change your backend](/agents)
- [Expose use cases as AI tools](/tools)
- [Gate AI behavior with evaluations](/evaluations)
- [Study the Fieldnotes reference backend](/fieldnotes)
---
# Let a coding agent change your backend
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 the applicable command:
`uv run tenchi make feature --dry-run --json`
`uv run tenchi make use-case --dry-run --json`
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 a durable job changed, run the job compatibility diff before
updating `jobs.json`. If an application tool changed, run the tool
compatibility diff before updating `tools.json`. If an evaluation policy
changed, run the evaluation compatibility diff before updating
`evaluations.json`.
For a contract-driven generated use case, create a change plan with `--plan`
and retain its reported `plan_id`. Finish with `uv run tenchi verify --base-ref
--change-plan --json` and report the receipt with the
files changed. Do not weaken `tenchi.toml` to make a failing requirement
disappear.
```
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 |
| `tenchi.toml` in every generated app | A protected, repository-owned declaration of the evidence required before a change is done |
| `.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 eval list --json` | Evaluation names, case names and schemas, thresholds, timeouts, and budgets without case inputs |
| `tenchi make ... --dry-run --json` | A mutation preview with the files and follow-up wiring steps before anything is written; contract-driven previews can include a prospective change plan |
| Contract-driven change plans | A record of the requested use case, its baseline, generated paths, and the conditions that must hold when it is finished; its [plan ID](/change-plans#verification-terms) is derived from its content |
| `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 jobs --json` | Registered durable job names and payload schemas without queued values |
| `tenchi check --json` | One bounded, complete result for formatting, linting, types, tests, architecture, and boundary snapshot drift |
| `tenchi verify --base-ref --json` | One [receipt](/change-plans#verification-terms) for the finished tree: source digest, baseline commit, policy requirements, checks, architecture, and all four compatibility reports |
| `tenchi preflight --json` | A redacted, timeout-bounded report 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 |
| Job compatibility commands | Proof that a new consumer still accepts durable messages created under the historical job contract |
| Tool compatibility commands | Proof that machine-facing input, output, errors, and safety did not break existing callers |
| Evaluation compatibility commands | Proof, without case inputs, that cases, metrics, thresholds, timeouts, budgets, and suite kinds did not silently weaken the release gate |
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, application tools, and evaluations. 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, evaluations, 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
uv run tenchi make use-case notes create_note \
--from-contract app.features.notes.contracts:create_note_contract \
--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.
When a contract exists, prefer `--from-contract`. The preview derives the
boundary parameters and return type from the loaded declaration, so an agent
does not have to transcribe the contract into a second signature. It also
rejects declarations that cannot be bound as routes or represented in OpenAPI.
After generation, `# tenchi: incomplete` markers and the generated failing test
keep `check` red until the agent implements and tests the behavior. Remove the
markers only when their placeholders have been replaced.
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.
When the requested structure itself should be verified, create the files and
plan together:
```shell
uv run tenchi make use-case notes create_note \
--from-contract app.features.notes.contracts:create_note_contract \
--plan .tenchi/changes/create-note.json \
--base-ref origin/main --json
```
Retain the returned `plan_id` outside the edited worktree, such as in the task
or orchestrator state. The final receipt reports the ID it verified. See
[verify a generated change](/change-plans) for the exact completion conditions and the
limits of this proof.
### 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, job-message, application-tool, or evaluation-policy 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 --diff openapi.json
uv run tenchi openapi --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
snapshot. See [Application tools](/tools) for the complete compatibility rules.
For an evaluation-policy change, compare the manifest too; it contains no case
inputs:
```shell
uv run tenchi eval snapshot --diff evaluations.json
uv run tenchi eval snapshot \
--diff-ref origin/main \
--snapshot evaluations.json
```
This catches removed cases or metrics, lower thresholds, larger budgets or
timeouts, reordered cases, and deterministic-to-model changes without running
providers or exposing case inputs. A missing historical snapshot is an error;
only a human-authorized first adoption should add `--allow-missing-baseline`,
and the result must record an `evaluation manifest baseline` metadata change.
See [AI evaluations](/evaluations) for the complete workflow.
For durable background messages, compare the job manifest before accepting its
snapshot:
```shell
uv run tenchi jobs --diff jobs.json
uv run tenchi jobs --diff-ref origin/main --snapshot jobs.json
uv run tenchi jobs --write jobs.json
```
Removing a job or narrowing the payloads its consumer accepts is breaking even
if the current producer no longer emits the old shape; stored messages may still
use it.
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
uv run tenchi verify --base-ref origin/main \
--change-plan .tenchi/changes/create-note.json --json
```
The receipt contains the resolved commit, current and historical
`tenchi.toml` requirements, complete check result, application summary,
diagnostics, unresolved relationships, and all four compatibility reports. It
passes only when the policy was not weakened, every enforced requirement
passes, the map has no diagnostics or unresolved relationships, and none of
the four boundaries contains a breaking or unknown change.
The `source` field identifies the exact application tree with its current
`HEAD`, SHA-256 digest, and clean-or-dirty state. Tenchi includes tracked and
nonignored untracked paths below the application root, then rechecks the
identity after each application-owned stage. If an agent, test, import, or
concurrent process leaves that tree changed at a verification checkpoint, the
receipt fails and must be rerun after the edits stop. Git environment variables
cannot redirect capture to another checkout. Ignored runtime artifacts stay
outside the identity.
When a change plan is supplied, the receipt additionally requires the planned
files, removed incomplete markers, exact registered contract and use case,
the accepted contract-derived signature, exact route bindings, and a direct
dependency from the exact generated test function. Keep that function's name
when replacing its failing body, retain its direct imported use-case binding,
and avoid decorators that replace the function. The plan also requires the
callable pytest collects to match that source definition, then requires at least
one invocation whose setup, call, and teardown all pass; skipped, xfailed,
xpassed, deselected, failed, errored, uncollected, or ambiguous results fail the
receipt. It also requires the plan's baseline commit to match the verification
baseline.
If an agent disables or removes a required stage in `tenchi.toml`, Tenchi
retains the historical requirement for that run and reports the policy change
as incompatible. The agent therefore cannot make a difficult check disappear
in the same change that weakens the policy. Without a policy file, Tenchi's
built-in policy requires check, architecture, and OpenAPI plus each optional
boundary whose composition module exists; generated applications commit the
policy so the repository owns its definition of done.
`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 JSON or human output
Every command in this loop has a `--json` mode that writes exactly one
versioned object to stdout, including on expected failures, so an agent can
branch on a stable `code` instead of parsing terminal prose. Human and JSON
modes describe the same operation: use JSON when another tool will make a
decision and human output when a person is reviewing in a terminal. The [CLI
reference](/cli) lists every command and the fields of its result.
## 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.
A change plan narrows that limitation but does not remove it. It proves that
the requested generated structure was completed and connected, and reports
planned-test execution from the project pytest process. Because project-owned
plugins and test code share that process, the result is not independent of the
repository's own code. It also cannot prove that the
implementation satisfies the product requirement or that its test assertions
are meaningful. Use externally owned hidden acceptance tests when the code
producer itself is outside the trust boundary.
Model evaluation execution is intentionally outside `check` and `verify`.
Those commands still check and compare the declared policy. Let an agent
inspect suites with `eval list` and `evaluation_diff`, but authorize `eval run`
separately because it may send application-owned case data to external
providers and incur cost. The
coding-agent MCP server follows the same rule: `evaluation_list` is available
by default, while `evaluation_run` requires `--allow-evaluation-runs`.
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 job and application-tool contracts, architecture
diagnostics, generator previews, OpenAPI, job, tool, and evaluation-policy
compatibility, deployment preflight, evaluation discovery, 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, job-message, application-tool, and evaluation-policy 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 |
| `jobs` | The registered durable job-message manifest with names and input schemas but no queued payloads |
| `doctor` | Source-anchored architecture and dependency diagnostics |
| `preflight` | Read-only, timeout-bounded observations of the target deployment environment with redacted results |
| `evaluation_list` | Evaluation names, case names and schemas, metric thresholds, timeouts, and budgets without case inputs |
| `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`; use-case previews accept `from_contract` to derive an exact boundary signature and `base_ref` to return an inline change plan |
| `openapi_diff` | A compatibility report against `openapi.json`, another project snapshot, or that snapshot at a Git ref |
| `jobs_diff` | A compatibility report against `jobs.json`, another project snapshot, or that snapshot at a Git ref |
| `tools_diff` | A compatibility report against `tools.json`, another project snapshot, or that snapshot at a Git ref |
| `evaluation_diff` | A compatibility report, without case inputs, against `evaluations.json`, another project snapshot, or that snapshot at a Git ref; missing Git snapshots fail unless first adoption is explicit, and evaluators never run |
| `verify` | One [receipt](/change-plans#verification-terms): source digest, optional change-plan result, `tenchi.toml` requirements, checks, architecture, and all four compatibility reports against a required Git ref |
| `check` | Ruff format, Ruff lint, Pyright, pytest, doctor, and the OpenAPI, job, tool, and evaluation-policy 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.
`evaluation_run` is also absent by default. Enable it only for a trusted
process that may call model providers and consume the application's declared
budget:
```shell
uv run tenchi mcp --allow-evaluation-runs
```
The result contains scores, usage, statuses, and stable failure codes without
case inputs, prompts, or model outputs. See [AI evaluations](/evaluations) for
runner composition, budget behavior, and the recommended deployment gate.
`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, job, tool, or
evaluation policy 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. When a use
case's HTTP contract already exists, pass its `module:attribute` target as
`from_contract` so the preview derives the exact boundary signature. Pass
the intended `base_ref` to receive a change plan inline, then retain its
`plan_id` outside the edited worktree.
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 `jobs_diff` before accepting a changed durable job-message snapshot.
8. Call `tools_diff` before accepting a changed application-tool snapshot.
9. Call `evaluation_diff` before accepting a changed evaluation-policy snapshot.
10. Call `verify` with the pull request base, previous push, or previous release.
If the CLI or another trusted tool persisted the inline plan, pass its
project-relative path as `change_plan`. Retain the completion receipt and
compare its `plan_id` with the initially accepted ID. Resolve policy changes
as well as failed stages; do not disable a `tenchi.toml` stage to make the
receipt pass.
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.
The MCP server never writes a returned change plan; persist it through an
authorized filesystem tool or create it with the CLI. See [verify a generated
change](/change-plans) for the exact pytest-target requirement, the completion
conditions, and what the receipt does and does not prove.
## 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 \
--evaluations my_app.server.evaluations:runner \
--tasks my_app.server.tasks:runner \
--jobs my_app.server.jobs:jobs \
--tools my_app.server.tools:tools \
--snapshot api/openapi.json \
--job-snapshot api/jobs.json \
--tool-snapshot api/tools.json \
--evaluation-snapshot api/evaluations.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`.
Job discovery and compatibility use `--jobs`; `jobs_diff`, `check`, and
`verify` use `--job-snapshot`.
Both `jobs_diff` and `verify` reject a missing historical job snapshot by
default. Use `allow_missing_baseline` on `jobs_diff`, or
`allow_missing_job_baseline` on `verify`, for the one first-adoption comparison.
The evaluation tools use `--evaluations`; `evaluation_diff`, `check`, and
`verify` use `--evaluation-snapshot`.
Both `evaluation_diff` and `verify` reject a missing historical evaluation
snapshot by default. A human may set `allow_missing_baseline` on
`evaluation_diff`, or `allow_missing_evaluation_baseline` on `verify`, for the
one first-adoption comparison. The `verify` override applies only when the
selected ref already contains the OpenAPI, job, and tool snapshots. The returned
policy result records an `evaluation manifest baseline` metadata change; do
not authorize this for a renamed or mistyped path.
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.
---
# Verify a generated change
Source: https://tenchi.io/change-plans
A change plan connects a contract-driven generator request to the final
`tenchi verify` receipt. Use one when a human or coding agent should prove that
the generated use case was implemented, tested, and wired to the intended HTTP
contract.
Change-plan version 2 supports use cases generated with `--from-contract`.
Plans are JSON files with a content-derived `plan_id`, the Git baseline, the
contract and use-case identities, generated paths, one exact pytest target, and
a fixed set of completion conditions.
## Verification terms
These words keep one meaning across `tenchi verify`, change plans, and the
coding-agent guides.
- **Baseline.** The Git commit that `--base-ref` resolves to. A ref such as
`origin/main` can move later, so the receipt records the commit and every
comparison uses that same historical state.
- **Receipt.** The result of `tenchi verify`: one JSON object covering the
source digest, verification policy, checks, architecture, compatibility
reports, and any change plan. `tenchi check` produces a result; only `verify`
produces a receipt.
- **Evidence.** What a `[verify]` stage in `tenchi.toml` requires, such as a
passing check or a compatible OpenAPI report. Each stage ends as `passed`,
`failed`, `skipped`, `not_configured`, or `not_verifiable`.
- **Source digest.** A SHA-256 hash of the application tree, including
nonignored untracked files, captured before verification and rechecked after
every project-owned stage.
- **Plan ID.** The `sha256:…` identity of a change plan, derived from its
content. Any change to the plan produces a new ID.
- **Payload-safe.** A result that never contains request bodies, case inputs,
prompts, model output, secrets, or exception text. Tenchi's manifests,
reports, and observer outcomes all hold to this.
- **Compatible.** A historical comparison passes when the new version still
accepts everything the old version accepted. The comparison is one-way: it
does not ask whether the two versions are equal.
## Preview the generation
Start from a committed baseline, then preview the files and boundary Tenchi
would generate:
```shell
uv run tenchi make use-case projects create_project \
--from-contract app.features.projects.contracts:create_project_contract \
--dry-run --json
```
The preview does not write application files. Review its derived signature and
planned paths before accepting the generation.
## Create the files and plan together
Run the generator without `--dry-run` and provide a project-relative plan path:
```shell
uv run tenchi make use-case projects create_project \
--from-contract app.features.projects.contracts:create_project_contract \
--plan .tenchi/changes/create-project.json \
--base-ref origin/main \
--json
```
Tenchi resolves `origin/main` to a baseline commit before writing anything.
The generated use-case file, failing test, and plan are created as one
filesystem operation; if any write fails, Tenchi removes the files created by
that operation.
The JSON result contains the complete plan, its destination, and a `plan_id`
such as `sha256:…`. Keep that ID in the task description, review record, or
orchestrator state. Any plan-content change produces a different ID, and the
final verification receipt reports the ID it actually checked.
Combining `--plan PATH` with `--dry-run` includes the prospective plan and
path in the result but writes neither the generated files nor the plan.
## Implement the behavior
Replace the generated `NotImplementedError` and failing test. Remove both
`# tenchi: incomplete` markers only after the behavior and its direct test are
implemented. Keep the generated test function name—`test_create_project` in
this example—because the plan binds completion evidence to that exact pytest
target. Bind the contract and use case explicitly in the feature's `routes.py`;
the generator does not rewrite composition modules.
Run the normal validation loop while working:
```shell
uv run tenchi check --json
```
## Verify the requested structure
Use the same baseline named in the plan:
```shell
uv run tenchi verify \
--base-ref origin/main \
--change-plan .tenchi/changes/create-project.json \
--json
```
The change-plan section passes only when:
- the plan and verification resolve to the same baseline commit;
- both generated files exist and contain no incomplete marker;
- the exact contract and use case are registered, and the current contract still
produces the accepted use-case signature;
- one registered route binds that contract and use case through exact
application-map evidence; and
- exactly one top-level test function matches the target, directly references
the imported use case, and neither the module nor function rebinds that
imported name;
- the callable collected by pytest comes from that exact source definition, so
a decorator that replaces or wraps the function fails as ambiguous; and
- pytest collects at least one invocation of that function, and every setup,
call, and teardown reports success. Skipped, expected-failure, unexpected
pass, deselected, failed, errored, uncollected, and ambiguous invocations all
fail the plan.
Tenchi normally records this evidence during the pytest step of `tenchi check`.
If the repository's current and historical verification policies both disable
the general check stage, Tenchi runs only the planned pytest target so the plan
cannot waive its own execution requirement.
Tenchi reads the plan before running project-owned validation commands and
again before returning the receipt. If a test or import mutates the plan during
verification, the receipt fails.
## What the receipt proves
A passing plan proves that the generated structure was completed and that its
planned test ran successfully in the project's own pytest process. It does not
prove business correctness, and nothing outside the repository attests the
run. The JSON receipt identifies this boundary as
`provenance: "project_pytest_process"`.
Project-owned `conftest.py` files, pytest plugins, fixtures, and imported test
code run in that same interpreter. Code with permission to modify and execute
the repository can therefore interfere with the evidence collector. Tenchi
clears inherited `PYTEST_ADDOPTS`, `PYTEST_PLUGINS`, and `PYTHONPATH` so shell
configuration cannot silently redirect the run, but it does not describe this
in-process evidence as tamper-proof. Use an external evaluator with withheld
tests—such as the repository's agent-change benchmark—or an isolated CI policy
owned outside the edited worktree when the code producer is adversarial.
The receipt also does not prove that the test called the use case, that its
assertions are strong enough, or that the implementation satisfies the product
requirement. Review the use case, policy, ports, and test behavior as usual;
`tenchi check` and the historical compatibility stages remain part of the same
final receipt.
Pytest marks such as `@pytest.mark.parametrize` remain suitable because they
preserve the test function's source identity. A custom decorator that returns a
different callable cannot provide plan-bound execution evidence.
Plans created before schema version 2 are not accepted. Generate a new plan
with the current Tenchi version before editing its files; do not hand-edit an
older plan into the new shape.
The plan file is repository-owned data, not a signed instruction. Tenchi
validates its content-derived identity and refuses weakened completion
conditions, but someone who can edit the repository can replace the whole
plan and receive a new ID. Preserve the initially accepted `plan_id` outside
the edited worktree when the exact requested intent needs independent review.
---
# Expose use cases as AI 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.
`ToolRunner` does not add a per-call timeout, and an HTTP contract's timeout
does not apply here. Put a shared application deadline in the use case or its
provider adapter when the operation must be bounded through HTTP, tools, and
MCP. See [set an entrypoint-neutral deadline](/execution#set-an-entrypoint-neutral-deadline).
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
```
Once `app/server/tools.py` exists, `tenchi check` compares the generated
manifest with `tools.json` so an unrecorded contract change fails locally and
in CI. Declare `tools = true` under `[verify]` in `tenchi.toml` in the same
change; `tenchi verify` rejects a composed boundary the policy omits, and
treats the missing historical `tools.json` as a first adoption when the
module did not exist at the baseline. 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 classifies changes against the historical manifest:
- 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 application 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]"
```
Tenchi uses the stable 2.x line of the official MCP Python SDK.
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, a detached, read-only copy of the request headers
with lowercase names. `McpRequest` representations omit that mapping, so
logging the request object does not disclose credentials. Resolve identity
from those headers, then build a runner whose context carries the authenticated
principal:
Replace `request.transport_request.headers` with `request.headers` and handle
`None` for stdio or direct in-process calls. Application MCP results now use
`schema_version: 2`; the transport-neutral `tools.json` manifest remains at
version 1.
```python
# app/server/mcp.py
from app.infra.tokens import token_directory
from app.server.runtime import DATABASE_PATH
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:
headers = request.headers
if headers is None:
raise PermissionError("This server requires HTTP authentication.")
scheme, _, token = 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(
database_path=DATABASE_PATH,
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`. Transport rules are listed under
[Constraints](#constraints).
When your process configures an OpenTelemetry provider, MCP v2 also emits
protocol-level telemetry. That complements Tenchi's payload-safe HTTP,
use-case, and tool outcomes; it does not replace application authorization or
Tenchi outcome observers.
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(
database_path=DATABASE_PATH,
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.
## 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 return ordinary MCP tool
results rather than JSON-RPC protocol errors. The results set MCP's `isError`
flag and retain structured details, so a client can inspect the reason, 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": 2,
"ok": true,
"result": {
"id": "project_123",
"name": "Launch"
}
}
```
Expected failures use the same envelope:
```json
{
"schema_version": 2,
"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.
Every `ok: false` result also sets MCP's standard `isError` flag. Generic MCP
hosts can use that signal to recognize an unsuccessful tool execution, while
clients that understand Tenchi can inspect the structured `error.kind`, stable
application error code, and safe details. These remain tool results rather than
JSON-RPC protocol errors.
`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 app.server.runtime import DATABASE_PATH
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(
database_path=DATABASE_PATH,
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.
For an integration test, create a server with a fixed test principal instead
of the HTTP authentication callback. After seeding one project visible to
`test_user` in `test_database_path`, run discovery and execution through an
in-memory MCP session:
```python
from mcp.client import Client
from tenchi.mcp import create_tool_mcp_server
test_mcp = create_tool_mcp_server(
tools=tools,
authenticate=lambda request: test_user,
runner_factory=lambda user: create_user_tool_runner(
database_path=test_database_path,
user=user,
),
)
async with Client(test_mcp) as client:
listed = await client.list_tools()
result = await client.call_tool(
"projects.search",
{},
)
assert [item.name for item in listed.tools] == ["projects.search"]
assert result.structured_content == {
"schema_version": 2,
"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.headers`.
## Constraints
- Application MCP supports the SDK's Streamable HTTP transport and stdio.
`stateless_http=False` is rejected because authentication and visibility are
evaluated independently for every request.
- The inherited `sse_app()` and `run_sse_async()` entrypoints are rejected.
Legacy SSE cannot carry an independently authenticated request-header scope
without weakening the identity and visibility guarantees on this page.
- `McpRequest.headers` is `None` for stdio and direct in-process calls, and its
representations omit the header mapping.
- Authentication or visibility failures during discovery return one generic
`Tool discovery failed.` MCP error. Callback exception messages and the
failed stage are not returned to the client.
- A tool declared with `destructive=True` cannot run without an `approve`
callback that accepts the exact call.
- `TOOL_MCP_PROTOCOL_VERSION` changes whenever the result envelope or schema
mapping changes. The transport-neutral `tools.json` manifest carries its own
version.
---
# Gate AI behavior with evaluations
Source: https://tenchi.io/evaluations
Application evaluations turn model behavior into an explicit release decision.
You declare typed cases, normalized metrics, minimum passing averages, and
optional token and cost budgets. Tenchi validates the declarations, runs each
case with your application lifecycle and context, and returns a payload-safe
pass/fail report.
Tenchi does not choose a model provider, prompt format, judge, agent loop, or
dataset store. Put those choices behind application-owned ports so the same
evaluation can compare providers or model versions.
## Declare typed cases and metrics
Keep an evaluation with the feature whose behavior it measures:
```python
# app/features/support/evaluations.py
from typing import Protocol
from pydantic import BaseModel
from tenchi.evaluations import (
EvaluationMeasurement,
evaluation,
evaluation_case,
evaluation_group,
evaluation_metric,
evaluation_result,
)
from .ports import AnswerGenerator
class AnswerCase(BaseModel, frozen=True):
question: str
required_fact: str
class EvaluationContext(Protocol):
answers: AnswerGenerator
async def evaluate_answer(
case: AnswerCase,
context: EvaluationContext,
) -> EvaluationMeasurement:
answer = await context.answers.generate(question=case.question)
required_fact_present = (
case.required_fact.casefold() in answer.text.casefold()
)
return evaluation_result(
scores={"required_fact_present": float(required_fact_present)},
tokens=answer.tokens,
cost_usd=answer.cost_usd,
)
answer_quality = evaluation(
"support.answer_quality",
case=AnswerCase,
cases=(
evaluation_case(
"support.answer_quality.refunds",
AnswerCase(
question="When can a customer request a refund?",
required_fact="30 days",
),
),
),
metrics=(
evaluation_metric(
"required_fact_present",
threshold=0.9,
description="Required policy facts appear in the answer.",
),
),
evaluator=evaluate_answer,
kind="model",
description="Measure whether required policy facts remain in support answers.",
timeout=20,
max_tokens=20_000,
max_cost_usd=2.0,
)
evaluations = evaluation_group(answer_quality)
```
`evaluation()` checks the evaluator's signature when its module imports. Naming
and value rules are listed under [Constraints](#constraints).
The context annotation may be a feature-owned `Protocol`. This keeps the
evaluation independent of server composition while allowing the concrete
`AppContext` to satisfy the same shape at runtime.
## Return scores, not generated payloads
`evaluation_result()` accepts only declared scores and optional usage:
```python
return evaluation_result(
scores={
"required_fact_present": required_fact_present,
"citation_quality": citation_quality,
},
tokens=usage.input_tokens + usage.output_tokens,
cost_usd=usage.cost_usd,
)
```
Do not put prompts, retrieved documents, model output, user data, or exception
text in score names, case names, metric descriptions, or evaluation
descriptions. Those declaration fields, and the case JSON Schema, are visible
through discovery.
Run-specific reports contain:
- evaluation and case names;
- completed, failed, timed-out, and skipped statuses;
- normalized scores and metric averages;
- declared thresholds;
- token and cost totals when reported;
- stable failure codes and durations.
Reports never contain case inputs, prompts, model outputs, context values, or
exception messages.
Name each metric after the property its scorer actually proves. A substring
check can establish that a required phrase is present; it cannot establish that
every generated claim is grounded in supplied evidence. Use a stronger name
such as `groundedness` only when the evaluator verifies that stronger property.
The CLI and coding-agent MCP server discard direct standard output and
standard error from evaluation code. They cannot redact handlers that send
data directly to files, telemetry, or a provider dashboard. Configure those
systems separately and never log prompts or model output unless your data
policy permits it.
## Compose the runner
Combine feature groups at the server composition root:
```python
# app/server/evaluations.py
from app.features.support.evaluations import evaluations as support_evaluations
from app.server.runtime import DATABASE_URL, create_context, create_lifespan
from tenchi.evaluations import create_evaluation_runner, evaluation_group
evaluations = evaluation_group(support_evaluations)
runner = create_evaluation_runner(
evaluations=evaluations,
context_factory=create_context,
lifespan=create_lifespan(DATABASE_URL),
concurrency=2,
)
```
One lifespan surrounds the complete run. Each case receives a separate scoped
context, so database transactions and other request-scoped resources still
commit, roll back, and close normally. Cancellation propagates through the
evaluator, context, and lifespan.
Cases run in declaration order with bounded concurrency. Reports retain that
order even when cases finish out of order. Evaluations themselves run
sequentially so each suite has an independent budget.
## Set thresholds, timeouts, and budgets
An evaluation passes only when:
- every case completes;
- every metric's average meets its threshold; and
- every declared usage budget is known and remains within its limit.
The declaration timeout applies to each case.
When you declare `max_tokens` or `max_cost_usd`, every completed measurement
must report the corresponding usage. Missing usage, a failed case, or a timed
out case leaves the budget unverified and stops later batches. Cases that were
already running may finish together, so a concurrent batch can exceed the
declared budget. Use `concurrency=1` when the budget must stop after each
individual case.
Budget outcomes report `passed`, `exceeded`, or `unverified`. Later cases use
`EVALUATION_BUDGET_EXCEEDED` only when measured usage crossed a limit and
`EVALUATION_BUDGET_UNVERIFIED` when the runner could not establish usage.
Cost totals are compared using exact decimal arithmetic, so a declared limit is
not crossed merely because of binary floating-point summation.
Budgets are gates, not provider-side spending limits. Configure provider
quotas and request limits separately.
## Protect the gate itself
Commit an evaluation-policy snapshot, which contains no case inputs, so a
source change cannot
silently weaken the release gate:
```shell
uv run tenchi eval snapshot --write evaluations.json
uv run tenchi eval snapshot --check evaluations.json
```
The snapshot includes evaluation names, case names in execution order, the case
JSON Schema, metric names and thresholds, suite kind, per-case timeout, and
token or cost budgets. It never includes case inputs and creating it never
invokes an evaluator. `tenchi check` performs the exact snapshot check once
`app/server/evaluations.py` exists; declare `evaluations = true` under
`[verify]` in `tenchi.toml` in the same change so `tenchi verify` enforces the
historical comparison.
Review policy changes before replacing the snapshot:
```shell
uv run tenchi eval snapshot --diff evaluations.json
uv run tenchi eval snapshot \
--diff-ref origin/main \
--snapshot evaluations.json \
--diff-format json
```
The compatibility report treats a stronger threshold, smaller timeout or
budget, and a newly added case or metric as additive. It fails for removed
evaluations, cases, or metrics; lower thresholds; larger or removed budgets;
larger timeouts; and changes from deterministic scoring to model scoring. Case
reordering, case schema changes, and unrecognized policy fields require review.
A missing snapshot at `--diff-ref` fails closed. During first adoption only,
authorize the absence explicitly and inspect the reported provenance:
```shell
uv run tenchi eval snapshot \
--diff-ref origin/main \
--snapshot evaluations.json \
--allow-missing-baseline \
--diff-format json
```
The result records an `evaluation manifest baseline` metadata change. Do not
use this override for a renamed or mistyped snapshot path. `tenchi verify`
needs no override when `app/server/evaluations.py` itself did not exist at the
baseline.
Case names are the stable identity in this manifest. Changing a case's input
while keeping its name does not change the snapshot. Review case
data in source control and rename the case when the scenario's meaning changes.
## Run the gate
Discover evaluations without running them:
```shell
uv run tenchi eval list
uv run tenchi eval list --json
```
Run every registered evaluation or select one stable name:
```shell
uv run tenchi eval run
uv run tenchi eval run support.answer_quality
uv run tenchi eval run support.answer_quality \
--concurrency 2 \
--timeout 15 \
--json
```
The default target is `app.server.evaluations:runner`. Override it with
`--evaluations module:attribute`.
The command exits zero only when every selected evaluation passes. Evaluation
execution stays separate from `tenchi check` and `tenchi verify`: those commands
verify the declared policy without contacting providers, while `eval run` may
be nondeterministic, depend on external services, and incur cost. Run it in a
deployment workflow that has the intended provider credentials and an explicit
budget.
## Let a coding agent inspect or run evaluations
The coding-agent MCP server always exposes `evaluation_list` and
`evaluation_diff`. Discovery returns case
names and schemas, metric thresholds, timeouts, and budgets without running a
provider or exposing case inputs. The diff tool compares the same manifest
used by the CLI and never runs an evaluator.
`evaluation_run` is disabled by default because it can call external systems
and spend money. Enable it only for a trusted MCP process:
```shell
uv run tenchi mcp --allow-evaluation-runs
```
MCP execution uses the same runner and result model as the CLI. Application
MCP tools created with `create_tool_mcp_server()` are a separate boundary; they
do not expose evaluation execution.
## Choose cases that make failures actionable
Use stable, reviewable cases for behaviors that matter to users:
- required facts and prohibited claims;
- tool-selection or routing decisions;
- structured-output validity;
- retrieval relevance;
- refusal and authorization boundaries;
- latency, token, and cost expectations.
Start with deterministic scorers where an exact rule exists. Use model judges
only for qualities that cannot be expressed reliably as code, and calibrate
their thresholds against examples a person has reviewed. Keep provider
requests idempotent or isolated when retries could incur duplicate cost.
`tenchi map` includes evaluation nodes with source locations, registration
state, suite kind, case count, metrics, timeout, and budgets. This gives coding
agents and reviewers one architecture view of both AI-facing tools and the
gates that protect their behavior.
## Constraints
- Case and evaluation names use dotted `snake_case`. Metric names use
`snake_case`.
- Every score is a finite number from `0` to `1`.
- The evaluator must be an async function with exactly annotated `case` and
`context` parameters that returns `EvaluationMeasurement`.
- The case JSON Schema is visible through discovery, including the field
titles, descriptions, defaults, and examples Pydantic produces. Tenchi
canonicalizes and validates that schema when the evaluation is declared, so
its metadata must be portable, standards-compliant JSON and must not contain
sensitive values.
- Token counts use JSON's interoperable integer range. `tokens` must be between
`0` and `MAX_EVALUATION_TOKENS` (`9_007_199_254_740_991`), and `max_tokens`
between `1` and that limit. Values outside the range are rejected before they
reach CLI or MCP output.
- `tenchi eval run --timeout` may shorten a declared case timeout but never
extend it.
- When `max_tokens` or `max_cost_usd` is declared, every completed measurement
must report the corresponding usage, or the budget is unverified.
---
# Study the cited AI reference backend
Source: https://tenchi.io/fieldnotes
Fieldnotes is a complete Tenchi application for saving research, indexing it
outside the request, searching owner-visible passages, and answering questions
with exact citations. It runs locally without model credentials and shows where
to connect a provider when you want generated synthesis.
The application lives in [`examples/fieldnotes`](https://github.com/taylorbryant/tenchi/tree/main/examples/fieldnotes).
## Run Fieldnotes
From the Tenchi repository:
```shell
cd examples/fieldnotes
uv sync
uv run tenchi check
uv run tenchi dev
```
The development API listens on `http://127.0.0.1:8000`, stores data in
`fieldnotes.db`, and recognizes two demonstration bearer tokens:
`alice-token` and `bob-token`.
The static tokens make the example immediately runnable. Replace
`StaticTokenDirectory` and the token map with your identity provider before
exposing the application beyond local development.
Start the indexing worker in another terminal:
```shell
cd examples/fieldnotes
uv run python -m app.server.worker
```
Saving a source returns `202` after committing both the source and a validated
`knowledge.index_source` outbox message. The worker splits the content into
passages, replaces any previous index entries, and marks the source indexed.
## Save and query material
Save text with an optional source URL:
```shell
curl http://127.0.0.1:8000/sources \
-H 'Authorization: Bearer alice-token' \
-H 'Content-Type: application/json' \
-d '{
"title": "MCP security",
"url": "https://example.com/mcp",
"content": "Explicit approval protects destructive MCP tools."
}'
```
Search after the worker has indexed the source:
```shell
curl http://127.0.0.1:8000/search \
-H 'Authorization: Bearer alice-token' \
-H 'Content-Type: application/json' \
-d '{"query":"destructive approval","limit":5}'
```
Ask a question with citations:
```shell
curl http://127.0.0.1:8000/answers \
-H 'Authorization: Bearer alice-token' \
-H 'Content-Type: application/json' \
-d '{"question":"What protects destructive tools?"}'
```
The answer includes a `has_citations` flag and citations containing the saved
source and exact passage ids. The flag reports whether citations are present;
it does not claim that generated text has passed a semantic grounding check.
With the included deterministic provider, a question with no matching evidence
returns a bounded insufficient-evidence answer with no citations. Authenticated
owner scope is applied to source listing, retrieval, and answering, so Bob
cannot discover Alice's material.
The starter accepts content plus an optional URL. It does not fetch arbitrary
URLs. Add a separately secured fetch adapter and allowlist appropriate for
your deployment before accepting remote locations from callers.
## Use the application tools
Fieldnotes exposes the same use cases as three application tools:
| Tool | Behavior |
| --- | --- |
| `knowledge.search` | Read-only, closed-world passage search |
| `knowledge.answer` | Read-only answer generation that may contact a configured provider |
| `sources.save` | Destructive source creation followed by background indexing |
Run the bearer-authenticated application MCP endpoint on a separate local port:
```shell
uv run uvicorn app.server.mcp:app --port 8001
```
Point an MCP client at `http://127.0.0.1:8001/mcp` and send
`Authorization: Bearer alice-token`. `create_fieldnotes_mcp_server()` resolves
that header for every discovery and invocation request, then creates a tool
runner carrying the authenticated user. Requests without a valid bearer token
fail authentication.
The included MCP application has no approval callback, so `sources.save`
returns `approval_required`. Supply an application-owned callback when creating
the server to approve an exact destructive call before its runner opens a
lifespan or transaction. The [application MCP guide](/tool-mcp) covers durable
approval and deployment transport security.
The project-local `.mcp.json` serves Tenchi's separate coding-agent MCP server;
it does not publish the Fieldnotes application tools.
## Connect a model provider
The `AnswerGenerator` protocol receives the question and already authorized,
owner-visible passages. It returns text, cited passage ids, and optional token
and cost usage. Implement that protocol in `app/infra/` using the model SDK you
choose, then select the adapter in `open_request_ports()`.
The answer use case gives the provider 25 seconds across HTTP, direct tool, and
MCP calls. Configure the SDK's own request timeout below that application
deadline so it can close network resources and preserve the original provider
diagnostic in approved telemetry.
The answer use case rejects duplicate citations and ids that were not supplied
to the provider. Keep provider credentials, prompts, raw model output, and
provider exceptions inside application-owned infrastructure and approved
telemetry systems.
The included `DeterministicAnswerGenerator` returns the highest-ranked passage
verbatim. It keeps the API, tools, and evaluation workflow usable without a
provider account.
After connecting a model, update `knowledge.answer_evidence` to declare
`kind="model"`. Keep a deliberate per-case `timeout` and `max_tokens`, add
`max_cost_usd`, and return both token and cost usage from the adapter. A missing
usage value makes its corresponding budget unverified rather than silently
passing.
Review the policy transition, run the model evaluation, and only then replace
the snapshot:
```shell
uv run tenchi eval snapshot --diff evaluations.json
uv run tenchi eval run knowledge.answer_evidence
uv run tenchi eval snapshot --write evaluations.json
uv run tenchi check
```
The first command fails closed because changing from deterministic to model
scoring requires review. That failure is expected until you accept the new
policy snapshot; an evaluation failure is not.
## Run quality and deployment gates
Run the deterministic retrieval and answer-evidence suites:
```shell
uv run tenchi eval list
uv run tenchi eval run
```
The evaluation policy requires the expected passage to rank first, required
evidence to remain in the answer, and every citation to name supplied evidence.
`evaluations.json` protects those cases, metrics, thresholds, and the token
budget from silent weakening.
Queue a reindex dry run or a real backfill through the validated operational
task:
```shell
uv run tenchi task run knowledge.reindex_sources \
--input '{"dry_run":true}'
```
Before deployment, initialize the database through the application lifespan,
then run:
```shell
uv run tenchi preflight
uv run tenchi check
```
Preflight opens SQLite read-only and verifies the source, passage, and outbox
tables. `check` verifies formatting, types, tests, architecture, and exact
OpenAPI, job, tool, and evaluation-policy snapshots.
---
# CLI reference
Source: https://tenchi.io/cli
The `tenchi` command scaffolds the prescribed structure, inspects the composed
application, and keeps API changes reviewable.
Commands with `--json`, plus compatibility commands using
`--diff-format json`, reserve stdout for one versioned JSON object. If Tenchi
cannot construct the command's normal result, it emits a shared
`operation_error` object with a stable operation and error code, then exits
nonzero. These failures omit application exception text and payloads. This
means an agent can parse both success and expected failure without falling back
to terminal prose.
## 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, an OpenAPI snapshot, a
`tenchi.toml` verification policy, a concise `AGENTS.md`, project-local MCP
configuration, and GitHub Actions CI. Nothing else is required to serve an
HTTP API.
```shell
uvx tenchi new my_app --full
```
`--full` also generates the composition modules for background jobs,
operational tasks, application tools, evaluations, and preflight checks, plus
the snapshot, snapshot test, and `tenchi.toml` stage for jobs, tools, and
evaluations, so every extension point exists from the start. Without it, add
each module when the application adopts that capability; the matching guide
shows the file, and `tenchi verify` treats the first snapshot as a first
adoption when the module did not exist at the baseline.
## 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
uv run tenchi make use-case notes create_note \
--from-contract app.features.notes.contracts:create_note_contract \
--dry-run --json
uv run tenchi make use-case notes create_note \
--from-contract app.features.notes.contracts:create_note_contract \
--plan .tenchi/changes/create-note.json \
--base-ref origin/main --json
```
Generators create files and print explicit wiring instructions. They never
rewrite route or infrastructure modules. A new feature receives `tasks.py`,
`jobs.py`, `tools.py`, or `evaluations.py` only when the matching
`app/server/` composition module exists, so a feature never gains a file that
nothing composes. `--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.
When the contract already exists, pass its `module:attribute` target through
`--from-contract`. Tenchi imports that declaration and derives the use case's
exact `params`, `query`, `headers`, `request`, context, and response
annotations in route-call order. Before planning files, it also proves that the
contract can be bound as a route and represented in OpenAPI, catching invalid
path parameters or unsupported boundary shapes at preview time. The target must
live in the selected feature's `contracts.py` module. Named boundary aliases
should live in that feature's `schemas.py` or `domain.py`, or under
`app.shared`, so the generated use case can import them without reversing the
application's dependency direction.
Name deeply nested boundary types there too; Tenchi fails early rather than
emit an inline annotation that would not pass the generated app's formatter.
Previewing from a contract imports that module to inspect its runtime types, so
keep contract modules declarative and free of startup I/O.
Contract-driven generation writes a `# tenchi: incomplete` marker in the use
case and its generated failing test. Ruff and Pyright can validate the boundary
immediately, while pytest and `tenchi doctor` keep `tenchi check` red until you
implement the behavior, replace the placeholder test, and remove both markers.
Contracts that use response definitions need an app-owned presenter result, so
Tenchi reports a structured configuration error instead of guessing that
return type. A contract with `response_headers` keeps its response type, and
the generated follow-up steps call out the required synchronous header
projector.
`--plan PATH` is available with `--from-contract`. It resolves `--base-ref`
(default `HEAD`) to a baseline commit and writes a versioned structural plan
in the same transaction as the generated use case and test. The result includes
the content-derived plan ID. With `--dry-run`, Tenchi returns the prospective
plan but writes neither the plan nor application files. See [verify a generated
change](/change-plans) for the complete workflow and proof boundary.
## 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,evaluation,use-case,policy,port --json
```
Map combines the canonical source layout with composed API routes, operational
tasks, background jobs, application tools, and evaluations. It returns a deterministic,
versioned graph covering features, contracts, routes, background jobs,
operational tasks, application tools, evaluations, 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. A default job, task, tool, or evaluation target whose module does not
exist is skipped; an explicitly overridden target must load. 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 default evaluation target is
`app.server.evaluations:runner`; override it with
`--evaluations 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 durable job messages
```shell
uv run tenchi jobs
uv run tenchi jobs --json
uv run tenchi jobs --diff jobs.json
uv run tenchi jobs --diff-ref origin/main --snapshot jobs.json
uv run tenchi jobs --check jobs.json
uv run tenchi jobs --write jobs.json
```
The default target is `app.server.jobs:jobs`; override it with
`--jobs module:attribute`. The manifest contains stable names, descriptions,
and producer-to-consumer input schemas without payloads or handler results.
`--check` detects exact drift. `--diff` and `--diff-ref` reject removed jobs,
narrower accepted payloads, and changes the analyzer cannot prove safe. Use
`--diff-format json` for a versioned compatibility result. See [Background
jobs](/jobs) for rollout rules around stored messages.
For first adoption only, `--allow-missing-baseline` permits a `--diff-ref`
whose commit predates `jobs.json` and records that fact as metadata.
## 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
against the historical baseline 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
uv run tenchi openapi --diff openapi.json
uv run tenchi openapi --diff-ref origin/main --snapshot openapi.json
uv run tenchi openapi --check openapi.json
uv run tenchi openapi --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.
Standalone `openapi`, `check`, and `verify` all default to
`app.server.routes:api_routes` and discover literal `OPENAPI_*` declarations
from that module. Common overrides are `--routes`, `--title`, `--version`,
`--description`, and `--security`.
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. Application source directories
must use real files and directories rather than symlinks; Doctor rejects links
instead of scanning different files on different Python versions.
## 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, job-message, application-tool, and evaluation-policy snapshot checks.
A job, tool, or evaluation snapshot step is omitted only when neither its
default composition module nor its snapshot file exists; a snapshot without its
module, or a module without its snapshot, still runs and fails as drift. 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`.
The job target defaults to `app.server.jobs:jobs`, and its snapshot defaults to
`jobs.json`. Override them with `--jobs` and `--job-snapshot`. The evaluation
target defaults to `app.server.evaluations:runner`, and its
snapshot defaults to `evaluations.json`. Override them with `--evaluations` and
`--evaluation-snapshot`.
## Verify a completed change
```shell
uv run tenchi verify --base-ref origin/main
uv run tenchi verify --base-ref origin/main --json
uv run tenchi verify --base-ref origin/main \
--change-plan .tenchi/changes/create-note.json --json
```
Verify produces one [receipt](/change-plans#verification-terms) for the finished source tree. It runs `tenchi
check`, rejects application-map diagnostics and unresolved relationships, and
compares the generated OpenAPI document, durable job-message manifest,
application-tool manifest, and evaluation-policy manifest with the snapshots at
the selected Git ref. The receipt records the baseline commit resolved from `--base-ref`, so all
four compatibility reports use the same historical state even if the named ref
moves later.
The receipt also records `source.head_commit`, `source.tree_digest`, and
`source.dirty` for the application root. The digest covers tracked files,
executable modes, symlink targets, tracked deletions, and nonignored untracked
files. Tenchi captures it before application-owned checks or imports, rechecks
it after every such stage, and fails with a `source` error if the tree remains
changed at a checkpoint. This prevents one receipt from combining checks run
against different source states. Git environment variables cannot redirect
capture to another checkout. Ignored runtime files do not affect the digest;
ignore local caches, databases, and logs that application commands are expected
to create.
When `--change-plan PATH` is present, the receipt also verifies the plan's
baseline, generated files, incomplete markers, exact contract and use-case
registration, the accepted contract-derived signature, route bindings, and a
direct dependency from the exact generated test function. Keep that function's
name when replacing its failing body. Verification requires pytest to collect
at least one invocation and report every invocation as passed; skipped,
xfailed, xpassed, deselected, failed, errored, uncollected, or ambiguous results
fail the receipt. The plan is read before and after project-owned commands so a
mid-verification mutation fails. The change-plan result supplements the normal
checks and compatibility reports; it does not evaluate business semantics or
test quality.
The application root may contain a repository-owned `tenchi.toml`:
```toml
schema_version = 1
[verify]
check = true
architecture = true
openapi = true
jobs = true
tools = true
evaluations = true
```
`true` makes the evidence required. `false` requests a deliberate skip, and an
omitted key records the current requirement as `not_configured`. A stage is
actually `skipped` or `not_configured` only when the historical policy did not
require it too; otherwise the historical requirement remains enforced and its
evidence reports `passed`, `failed`, or `not_verifiable`. Generated
applications declare `check`, `architecture`, and `openapi`; `tenchi new
--full` declares all six. A stage omitted from `tenchi.toml` while its
composition module exists is a verification error, so adopting a capability
means declaring its stage in the same change; `false` remains a deliberate
skip. An application without `tenchi.toml` uses
Tenchi's built-in policy: `check`, `architecture`, and `openapi` are required,
and `jobs`, `tools`, and `evaluations` are required while their default
composition module exists and `not_configured` otherwise. The current policy
judges the working tree; the historical policy judges the baseline commit, so
deleting a module cannot quietly drop a requirement the baseline had.
`check` means the complete local `tenchi check` loop, including exact drift
checks for every checked-in snapshot. `openapi`, `jobs`, `tools`, and
`evaluations` mean their separate historical compatibility comparisons against
the selected Git baseline. Disabling one historical comparison does not remove
its exact snapshot check while `check` remains required.
Verify compares the current policy with `tenchi.toml` at the resolved baseline.
Changing a required stage to `false` or removing its key is incompatible. Tenchi
still runs a stage required by either policy, so a change cannot disable the
check that would report its own weakening. Adding a repository policy is safe
only when it preserves the built-in requirements; removing a committed policy
fails closed. Invalid TOML, unknown stages, unsupported schema versions, and
unreadable historical policies also fail without suppressing the strict
built-in stages. Tenchi reads the policy again after every project-owned stage
has completed and rejects the receipt if a test or import changed it while
verification was running.
The structured receipt gives every stage its current and historical
requirement, whether it was enforced, and one of `passed`, `failed`, `skipped`,
`not_configured`, or `not_verifiable`. `policy.ok` is true only when the policy
is compatible and all enforced evidence passed.
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.
When a job, tool, or evaluation composition module did not exist at the
baseline, verify treats the missing historical snapshot as a first adoption
automatically and records a `job manifest baseline`, `tool manifest
baseline`, or `evaluation manifest baseline` metadata change. The overrides
below cover a module that did exist at the baseline while its snapshot did
not, such as a renamed snapshot path. If the selected ref genuinely predates
the application's first `jobs.json` in that situation, pass
`--allow-missing-job-baseline` once and confirm the job result includes a
`job manifest baseline` metadata change. If it already contains the OpenAPI,
job, and tool snapshots but genuinely predates the application's first
`evaluations.json`, pass
`--allow-missing-evaluation-baseline` once and confirm the evaluation result
includes an `evaluation manifest baseline` metadata change. Missing evaluation
snapshots otherwise fail closed, so a renamed or mistyped path cannot silently
replace the historical policy.
`--snapshot`, `--job-snapshot`, `--tool-snapshot`, `--evaluation-snapshot`, and the route, task,
job, tool, evaluation, and OpenAPI metadata options override the generated
application conventions. Verify never updates snapshots. Because it runs the
application's tests and validation commands, those commands retain their normal
side effects and per-step timeout.
Verify loads `app.server.evaluations:runner` for the architecture check and the
policy comparison but does not run evaluators. Use `--evaluations
module:attribute` when the application uses another composition target.
## 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 AI evaluations
```shell
uv run tenchi eval list
uv run tenchi eval list --json
uv run tenchi eval snapshot --diff evaluations.json
uv run tenchi eval snapshot --diff-ref origin/main \
--snapshot evaluations.json
uv run tenchi eval snapshot --check evaluations.json
uv run tenchi eval snapshot --write evaluations.json
uv run tenchi eval run
uv run tenchi eval run support.answer_quality --json
```
`eval list` discovers `app.server.evaluations:runner` and returns case names
and schemas, metrics, thresholds, timeouts, and token or cost budgets without
case inputs. `eval snapshot` prints, writes, checks, or compares the policy, which contains
no case inputs, with a historical snapshot without invoking evaluators. Breaking and unknown
changes return a non-zero status; `--diff-format json` returns the versioned
`evaluation_diff` result. A missing Git snapshot fails by default. During first
adoption only, pass `--allow-missing-baseline` and require the resulting
`evaluation manifest baseline` metadata change; do not use it for a renamed or
mistyped path. `eval run` opens the application lifespan, gives each case a
scoped context, and exits non-zero when a case, threshold, or budget fails.
Use `--concurrency` to override the number of cases in flight and `--timeout`
to tighten each case's declared timeout. Use `--evaluations module:attribute`
to override the runner target. Evaluation execution stays separate from
`check` and `verify` because it may call external models, vary between runs,
and incur cost. Those commands verify only the declared policy. See [AI
evaluations](/evaluations) for declarations, scoring, redaction, and deployment
guidance.
## 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`, `--evaluations`, `--tasks`,
`--jobs`, `--tools`, `--snapshot`, `--job-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.
Pass `--allow-evaluation-runs` only when the connected agent may call providers
and consume evaluation budgets; evaluation 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.
---
# Python 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 these convenience names. `__version__` reports the
installed package version:
```python
from tenchi import (
EVALUATION_MANIFEST_VERSION,
IDEMPOTENCY_CONFLICT,
IDEMPOTENCY_IN_PROGRESS,
JOB_MANIFEST_VERSION,
MAX_EVALUATION_TOKENS,
RATE_LIMITED,
TOOL_MANIFEST_VERSION,
AppError,
Client,
ClientAttemptObserver,
ClientAttemptOutcome,
ClientObserver,
ClientOutcome,
ClientResponse,
ConfigurationError,
Contract,
ErrorDef,
Evaluation,
EvaluationBindingError,
EvaluationBudgetOutcome,
EvaluationBudgetStatus,
EvaluationCase,
EvaluationCaseOutcome,
EvaluationCaseStatus,
EvaluationGroup,
EvaluationKind,
EvaluationManifest,
EvaluationManifestEntry,
EvaluationMeasurement,
EvaluationMetric,
EvaluationMetricManifest,
EvaluationMetricOutcome,
EvaluationNotFoundError,
EvaluationOutcome,
EvaluationReport,
EvaluationResultError,
EvaluationRunner,
ExecutionError,
ExecutionInputError,
IdempotencyResultError,
IdempotencyStore,
IdempotencyStoreError,
Job,
JobBindingError,
JobDispatcher,
JobGroup,
JobHandler,
JobManifest,
JobManifestEntry,
JobMessage,
JobNotFoundError,
JobPayloadError,
JobResultError,
MemoryIdempotencyStore,
MemoryRateLimitStore,
OutcomeObserver,
Page,
PageQuery,
PreflightBindingError,
PreflightCheck,
PreflightGroup,
PreflightOutcome,
PreflightReport,
PreflightStatus,
PresentedResponse,
RateLimitExceeded,
RateLimitPermit,
RateLimitStore,
RateLimitStoreError,
RequestInfo,
RequestOutcome,
ResponseDef,
RetryPolicy,
RetryTimeoutError,
Route,
RouteGroup,
Task,
TaskBindingError,
TaskGroup,
TaskInputError,
TaskNotFoundError,
TaskResultError,
TaskRunner,
TenchiError,
Tool,
ToolBinding,
ToolBindingError,
ToolGroup,
ToolInvocationError,
ToolNotFoundError,
ToolResultError,
ToolRunner,
UnexpectedResponseError,
UseCaseObserver,
UseCaseOutcome,
Webhook,
WebhookBindingError,
WebhookRequest,
WebhookVerifier,
__version__,
contract,
create_app,
create_evaluation_runner,
create_job_dispatcher,
create_task_runner,
create_tool_runner,
execute,
enforce_rate_limit,
evaluation,
evaluation_case,
evaluation_group,
evaluation_manifest,
evaluation_metric,
evaluation_result,
fingerprint,
health_route,
job,
job_group,
job_handler,
job_manifest,
job_message,
openapi_route,
openapi_schema,
page,
preflight_check,
preflight_group,
present,
response,
retry_policy,
route,
route_group,
run_idempotently,
run_preflight,
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`, `ExecutionInputError`, `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`, `JobManifest`, `JobManifestEntry`, `JOB_MANIFEST_VERSION`, `JobHandler`, `JobGroup`, `JobDispatcher`, `JobBindingError`, `JobNotFoundError`, `JobPayloadError`, `JobResultError`, `job()`, `job_message()`, `job_handler()`, `job_group()`, `job_manifest()`, `create_job_dispatcher()` |
| `tenchi.evaluations` | `EVALUATION_MANIFEST_VERSION`, `EvaluationManifest`, `EvaluationManifestEntry`, `EvaluationMetricManifest`, `Evaluation`, `EvaluationCase`, `EvaluationMetric`, `EvaluationMeasurement`, `EvaluationGroup`, `EvaluationRunner`, `EvaluationCaseOutcome`, `EvaluationCaseStatus`, `EvaluationMetricOutcome`, `EvaluationBudgetOutcome`, `EvaluationBudgetStatus`, `EvaluationKind`, `EvaluationOutcome`, `EvaluationReport`, `EvaluationBindingError`, `EvaluationNotFoundError`, `EvaluationResultError`, `MAX_EVALUATION_TOKENS`, `evaluation()`, `evaluation_case()`, `evaluation_metric()`, `evaluation_result()`, `evaluation_group()`, `evaluation_manifest()`, `create_evaluation_runner()` |
| `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`, `TaskBindingError`, `TaskInputError`, `TaskNotFoundError`, `TaskResultError`, `task()`, `task_group()`, `create_task_runner()` |
| `tenchi.tools` | `TOOL_MANIFEST_VERSION`, `Tool`, `ToolBinding`, `ToolGroup`, `ToolRunner`, `ToolManifest`, `ToolManifestEntry`, `ToolAnnotationManifest`, `ToolErrorManifest`, `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. An `EvaluationRunner`
applies the same lifecycle discipline to typed, budgeted AI quality gates
without choosing a model provider or judge.
## 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_job_compatibility()`, `analyze_tool_compatibility()`, `analyze_evaluation_compatibility()`, `render_compatibility_report()`, `render_job_compatibility_report()`, `render_tool_compatibility_report()`, `render_evaluation_compatibility_report()` |
| `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,
job-message, application-tool, and evaluation-policy 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.
---
# Upgrade Tenchi safely
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
Tenchi gives each public surface an explicit compatibility boundary:
- Public Python signatures are checked as part of every release. Intentional
changes appear in the changelog so you can identify required application
updates before upgrading.
- OpenAPI, durable job messages, application tools, and evaluation policy have
compatibility reports you can run against your own historical snapshots.
- Structured CLI results and the coding-agent MCP server share one versioned
schema. Application-tool manifests and application MCP results use separate
versioned schemas. Contract-driven change plans also carry their own schema
version and a content-derived plan ID. A breaking wire-format change receives a new
version rather than silently changing an existing one.
- `tenchi.toml` protects the evidence your repository requires before a change
is complete. `tenchi verify` rejects a change that weakens a requirement
relative to the selected historical commit.
These guarantees make changes visible and reviewable. 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. Before running
`tenchi verify` for its [receipt](/change-plans#verification-terms), update `.gitignore` for caches, logs,
coverage output, and local databases that project commands are expected to
create. Do not ignore source, snapshots, lockfiles, or configuration.
3. Run `tenchi check` to verify formatting, lint, types, tests, architecture,
and the current OpenAPI, job-message, application-tool, and
evaluation-policy snapshots.
4. Compare OpenAPI with the same route target and metadata used to create the
snapshot:
```shell
uv run tenchi openapi --diff openapi.json
```
The command discovers the route module's literal `OPENAPI_*` declarations;
pass explicit metadata flags only when overriding that convention.
5. If the application has durable jobs, run `tenchi jobs --diff jobs.json`
against the pre-upgrade snapshot.
6. If it has application tools, run `tenchi tools --diff tools.json` against
the pre-upgrade snapshot.
7. If it has evaluations, run `tenchi eval snapshot --diff evaluations.json`
against the pre-upgrade snapshot.
8. Review application wiring, `tenchi.toml`, and generated-code convention
changes. Do not weaken verification requirements to make the upgrade pass.
9. Update snapshots only after accepting all compatibility reports.
10. Rerun `tenchi check` with the accepted snapshots.
11. 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.
---
# Choose the right framework
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.