soev_agents.entrypoints.http

Make the FastAPI app that serves agents over store.

The app serves the threads API at the routes of mount. With the default mount, the threads API is under /threads, with GET /v1/capabilities and GET /healthz. The hooks apply to the routes of the threads API, except the health route.

With responses, the app also serves the OpenAI Responses API under /v1/responses, and GET /v1/models. The Responses API is experimental. It calls no hook, so it does not apply the server's access rules. Use it only for a server without access rules. resources applies to both APIs.

The four functions list_threads, list_tools, list_models and complete_task serve the optional routes of the public profile (Mount.public()). A route that mount names, but whose function is None, responds with 501. When mount.models is /v1/models, the threads API serves that route, also with responses.

The app maps the runtime's refusals to status codes, as DefaultHooks.error does. An input that the agent refuses, or a log position outside the thread, gets 422. A fresh input while a turn has not ended, a cancel with nothing to cancel, a thread id in use, or a thread that another process runs, gets 409.

At shutdown, the app waits for the running turns, then closes the observer, then leaves lifespan. So the server's clients stay open while the turns finish.

from soev_agents.adapters import InMemoryTraceStore
from soev_agents.entrypoints.http import ListQuery, Mount, RequestContext, ThreadList, create_app
from soev_agents.toolkit import ReActAgent

# config is a ReActConfig of your own.


async def list_threads(ctx: RequestContext, query: ListQuery) -> ThreadList:
    # `index` is the server's own index of threads, which its hooks fill
    return ThreadList(data=await index.threads_of(ctx.identity.subject, limit=query.limit))


app = create_app(
    InMemoryTraceStore(),
    {"assistant": ReActAgent(config)},
    version="1.4.0",
    mount=Mount.public(),
    list_threads=list_threads,
)

Run it with any ASGI server, for example uvicorn.

Raises:

• ValueError: keepalive is not a positive, finite number.

The version of the threads API contract that this code serves.

GET /healthz reports it, and the OpenAPI document carries it as its version. A client reads it to check that it speaks the same contract as the server. The contract is at 0.x, so names and shapes can still change.

An async function that returns the resources of an agent for one HTTP request.

The app calls it with the HTTP request and the agent each time that it builds a graph, never for a read, a cancel or a fork. Its result goes to the agent's build as resources, in a type that the server chooses. Use it to give each agent the clients of the caller, for example a model client with the caller's credentials. The function gets the HTTP request, not the turn's input: the input goes to the agent only.

Give no resources: the default resources of create_app, for agents that need none.

The routes of the threads API that a deployment serves, and how they behave.

The defaults serve the threads below /threads, with the status route and without the optional routes. Mount.public() gives the routes of the public profile. The conformance suite reads the same fields to find the routes that it checks.

The app of create_app serves the routes of its mount argument. The default is Mount(thread_parameter="thread_id", capabilities="/v1/capabilities", health="/healthz").

The server's hooks on each HTTP request of the threads API: authentication, authorization and errors.

The framework has no access-control model. The hooks give the means, and the server gives the rules. The app calls the hooks of one HTTP request in this order:

1. subject finds the caller.
2. authorize allows or refuses the call.
3. extend can change the call.
4. before_dispatch runs just before the operation.
5. The operation runs. When it fails, on_failed runs.
6. on_accepted runs when the operation accepted the call, and on_opened after it for a new thread. For open, inputs and resume, the turn has not run yet: it runs while the app sends the response.
7. For an event stream, on_frame runs for each frame, and stream_error renders a failure of the stream.

An error in steps 1 to 6 goes to error, which renders the HTTP response. A hook refuses a call when it raises, for example fastapi.HTTPException with a status code.

Subclass DefaultHooks and override only the hooks that you need.

from fastapi import HTTPException, Request

from soev_agents.entrypoints.http import DefaultHooks, Identity, create_app

# USERS maps each bearer token to the id of its user. store is a TraceStore, and agents maps each
# agent name to an Agent of your own.


class TokenHooks(DefaultHooks):
    async def subject(self, request: Request) -> Identity:
        token = request.headers.get("authorization", "").removeprefix("Bearer ")
        if token not in USERS:
            raise HTTPException(401, "unknown token")
        return Identity(USERS[token])


app = create_app(store, agents, version="1.4.0", hooks=TokenHooks())

The hooks of a server without authentication: each caller is anonymous, and each call is allowed.

create_app uses them when it gets no hooks. Subclass them and override only the hooks that you need. Each hook that is not listed below does nothing.

• subject returns an anonymous identity.
• extend returns the call unchanged.
• error renders the errors of the framework and fastapi.HTTPException, and raises each other error again.
• stream_error raises the error again, so the stream fails without a last frame.

Who makes an HTTP request, as Hooks.subject finds it.

One HTTP request of the threads API and its caller, as each hook receives them.

One call of the threads API: the operation, its thread and its parsed body.

The hooks receive the call before it runs. Hooks.extend can return a changed copy, for example with dataclasses.replace.

A call that is ready to run, as the hooks after extend receive it.

The app of create_app runs each call in its own process, so only call holds information there. The other fields serve a backend that sends the call to another server.

The result of one operation, before the app sends it as the HTTP response.

Hooks.on_accepted receives it. A complete_task function of create_app gives one, inside a ReplyContext.

The name of one operation of the threads API, as a hook receives it in Call.operation.

With the default mount of create_app, the app serves these operations:

• open: POST /threads
• inputs: POST /threads/{thread_id}/inputs
• resume: POST /threads/{thread_id}/resume
• cancel: POST /threads/{thread_id}/cancel
• fork: POST /threads/{thread_id}/fork
• read: GET /threads/{thread_id}
• export: GET /threads/{thread_id}/export
• status: GET /threads/{thread_id}/status
• events: GET /threads/{thread_id}/events
• delete: DELETE /threads/{thread_id}
• capabilities: GET /v1/capabilities

list, tools, models and task name the optional routes of a Mount: GET <threads> with list_threads, and the paths tools, models and tasks. Mount.public() serves all four.

The HTTP response to an error, as Hooks.error renders it.

The last frame of an event stream that failed, as Hooks.stream_error renders it.

The status code is sent when the stream starts, so a later error cannot change it. The app sends this frame instead, as event: <event> and data: <JSON of data>, and then ends the stream.

The body of a Call that the hooks get, by operation.

open has an OpenThread, inputs a SendInput, fork a ForkThread, and task a TaskRequest. cancel has a CancelThread, or None when the client sent no body. Each other operation has None. Check the type with isinstance before you read a field.

What it takes to open a thread: the agent, your first input, and optionally the id you want.

It is the body of POST /threads, and a hook finds it in Call.body for the open operation.

What it takes to continue a thread: the input, and the request it answers if it answers one.

It is the body of POST /threads/{thread_id}/inputs, and a hook finds it in Call.body for the inputs operation.

What it takes to fork a thread: the position of the last event to copy.

It is the body of POST /threads/{thread_id}/fork, and a hook finds it in Call.body for the fork operation.

What it takes to cancel the turn of one input, instead of the current turn.

It is the optional body of POST /threads/{thread_id}/cancel. A hook finds it in Call.body for the cancel operation, or None when the client sent no body.

An async function that gives the thread list of a caller, the list_threads argument of create_app.

The app calls it for GET <threads> after the hooks allowed the call, with the HTTP request, its caller and the query. The app keeps no thread list itself: the server keeps one, for example in Hooks.on_opened. An error goes to Hooks.error.

The query of a thread list, from the query string of GET <threads>.

The app validates the query string before the list function receives it, and refuses an invalid one with 422. The app does not apply the query: the list function does.

An async function that gives the tool list for a caller, the list_tools argument of create_app.

The app calls it for the route Mount.tools, after the hooks allowed the call. An error goes to Hooks.error.

An async function that gives the model list for a caller, the list_models argument of create_app.

The app calls it for the route Mount.models, after the hooks allowed the call. The list does not change which models an input can name: the models argument of create_app does. An error goes to Hooks.error.

A function that runs one task completion, the complete_task argument of create_app.

The app calls it for POST <Mount.tasks>, after the hooks allowed the call, with the validated body. A task completion is one model call in the format of the OpenAI Chat Completions API, outside any thread. The app records nothing in a thread for it. The function returns a ReplyContext. Make its reply with task_completion, or, when the body sets stream, as a text/event-stream of task_chunk frames.

An async context manager that gives the BackendReply of one call.

The app enters it and reads the reply. For a JSON reply, the app leaves it after it reads the body, before it sends the response. For a stream, the app leaves it after the last frame, also when the client disconnects. So the context manager can close what the stream reads, for example an HTTP response from a model server. An error before the reply goes to Hooks.error.

The body of a thread list, which a list_threads function of create_app gives.

One thread in a thread list: its id, its agent, when it was opened, and its title.

A list_threads function of create_app gives these in the data of a ThreadList.

The body of the tool list, which a list_tools function of create_app gives.

One tool in the tool list: its name, and how a client shows a call of it.

The body of the model list, which a list_models function of create_app gives.

One model in the model list: what a client shows to people, and the id that an input names.

What a model can do, each as a CapabilityState.

Whether a model has a capability: supported, unsupported, or unknown when nobody stated it.

How a model gives output in a JSON schema.

native: the model takes the schema in the request. json_mode: the model takes a request for JSON, and the instructions name the schema. unsupported: the model gives no structured output. unknown: nobody stated it.

A text that a client shows to people, in Dutch and in English.

Where the endpoint of a model runs, who runs it, and how long it keeps the data of a call.

The body of a task completion, in the OpenAI Chat Completions format, without a model.

The app validates it and gives it to the complete_task function of create_app. Validation refuses a field that the class does not declare.

One message of a task completion, in the OpenAI Chat Completions format.

Validation refuses a message that breaks these rules:

• The roles system, developer and assistant take text content only.
• Each message has content, except a message of the role assistant with tool_calls.
• Only the role assistant has tool_calls.
• The role tool has a tool_call_id, and only that role has one.

The content of a TaskMessage: a text, or a list of text and image parts.

A text part of the content of a TaskMessage.

An image part of the content of a TaskMessage.

The address of an image in an ImageContent.

A tool call of the model, in a TaskMessage of the role assistant or in a TaskAnswer.

The function of a HistoryToolCall: the name of the tool and its arguments.

The response_format of a task completion that asks for text output.

The response_format of a task completion that asks for a JSON object, without a schema.

The response_format of a task completion that asks for JSON output in a schema.

The JSON schema of a SchemaFormat.

Make the JSON body of a task completion that does not stream, in the OpenAI Chat Completions format.

The body is a chat.completion object with one choice, at index 0, whose message is answer. The message has the role assistant, and it leaves out reasoning_content when it is None, and tool_calls when it is None or empty. Give the body to the app in a BackendReply with the media type application/json.

Make one frame of a task completion that streams, in the OpenAI Chat Completions format.

The frame is a server-sent event without an event name: data: <JSON> and a blank line. Its JSON is a chat.completion.chunk object with one choice, at index 0, that holds delta. With usage, the frame has no choice and gives the token counts only. This function does not make the data: [DONE] frame that ends an OpenAI stream. Send data: [DONE] yourself as the last frame.

The message that the model gave for a task completion, as task_completion takes it.

The token counts of the model call of a task completion.

One thread's stored events, every field, in log order: a trace file.

It is the body of GET /threads/{id}/export. Unlike a read, it leaves out nothing that only the model sees, such as the images of a tool result. The app reads it from the store directly, so a thread whose agent is gone exports too.