Call contracts with the typed 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

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.

Inspect the HTTP response

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

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:

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:

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 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:

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:

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

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:

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.
• 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.