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

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",),
    base_delay_seconds=0.2,
    max_delay_seconds=2,
    total_timeout_seconds=5,
)

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

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

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

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

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.

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

Observe outbound calls

Pass synchronous or asynchronous observers when you construct the client:

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.