Run use cases outside HTTP

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. Jobs, tasks, and tools use the same request and context convention.

Adapt HTTP inputs

After completing Build a feature, the complete_todo function accepts params and context. To expose it as a task, add a function that receives the same data through request:

python

# app/features/todos/use_cases/complete_todo_from_request.py
from app.server.context import AppContext

from ..schemas import CompleteTodoParams, Todo
from .complete_todo import complete_todo


async def complete_todo_from_request(
    request: CompleteTodoParams,
    context: AppContext,
) -> Todo:
    return await complete_todo(params=request, context=context)

Bind the adapter in app/features/todos/tasks.py:

python

from tenchi.tasks import task, task_group

from .use_cases.complete_todo_from_request import complete_todo_from_request


tasks = task_group(
    task("todos.complete", complete_todo_from_request),
)

Compose this group with the operational task runner. The adapter can also receive validated input through execute() or a matching job or tool declaration. Business logic and authorization still run in complete_todo. If several HTTP inputs are needed, define a request model that contains them and unpack it in the adapter.

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 and Retries and background work. 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 to add stable discovery plus result validation around the same use cases.