Agent Server API

kavalai.server serves a workflow over HTTP. A router built from a WorkflowEngine exposes the workflow’s own input and output types as the request and response schemas, so the endpoints are typed by the graph rather than by hand:

Endpoint

What it does

POST /run_agent

Runs the workflow and returns the final output in one response.

POST /stream_agent

Runs the workflow and streams progress as Server-Sent Events; each frame is a WorkflowStreamEvent.

GET /workflow

Returns the workflow graph (used by the backoffice to render it).

GET /liveness / GET /health

Liveness and readiness (readiness also checks the database).

Both run endpoints accept an optional session_id to continue a conversation, or an external_id to let the caller key a session by its own identifier. HTTP basic auth is optional and configured from the environment (KAVALAI_AGENT_BASIC_AUTH_USER / KAVALAI_AGENT_BASIC_AUTH_PASSWORD).

Streaming a run over SSE

POST /stream_agent drives run_stream() and renders each event as an SSE frame — event: carries the event type and data: its JSON payload. A : ping comment frame is emitted during silent stretches (a long tool call, say) so proxies do not drop the connection.

Two consequences of SSE are worth planning for. A response cannot change its status code once the headers are sent, so a failed run ends the stream after a workflow_failed event instead of returning an error status — clients must treat that event as the failure signal. And disconnecting aborts the run: closing the stream cancels the engine generator, which records the abort on the run row.

Because the endpoint is a POST with a JSON body, browsers cannot consume it with EventSource (which supports neither a request body nor auth headers) — use fetch() with a streaming reader. From Python, use stream_agent(), which yields each data: payload as it arrives:

from pydantic import BaseModel

from kavalai.client import AgentClient


class Message(BaseModel):
    user_message: str


client = AgentClient("http://localhost:8000")
async for chunk in client.stream_agent(Message(user_message="Hi there")):
    print(chunk)

See WorkflowStreamEvent for the event contract and Workflows for which nodes emit content events.

Server

Launch Kaval.AI agent REST server.

SSE_PING_INTERVAL_SECONDS is how long a stream may stay silent before a keepalive comment is sent. SSE_HEADERS are the response headers every SSE stream carries: no caching, and no buffering by a reverse proxy. PUBLIC_FAILURE_MESSAGE is what public_events() sends in place of a failed run’s error text.

Copyright 2026 OÜ KAVAL AI (registry code 17393877)

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

class kavalai.server.AgentRequest(*, session_id: UUID | None = None, external_id: str | None = None, data: InputT)[source]

Bases: BaseModel, Generic[InputT]

The request body of POST /run_agent and POST /stream_agent.

data is the workflow’s own input type. session_id continues a session by its id; external_id continues — or starts — the session carrying the caller’s own key. A host that serves a workflow from a route of its own types the body as AgentRequest[MyInput], so its clients speak the same protocol as the SDK’s router.

session_id : UUID | None
external_id : str | None
data : InputT
model_config : ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

kavalai.server.public_events(event: WorkflowStreamEvent) → WorkflowStreamEvent | None[source]

An event filter for streams served to people other than the operator.

Forwards the run lifecycle and the streamed content, and withholds what only the operator should see:

  • node_started / node_completed events, which describe the workflow’s internals;

  • token usage;

  • the reason attached to a restart;

  • the error text of a failed run, which can carry a provider’s message. It is replaced by PUBLIC_FAILURE_MESSAGE and the run id, so a report can be matched to the recorded run; the detail stays in the server log and on the run row.

Which nodes stream content at all is the workflow’s stream_output setting, not this filter’s.

kavalai.server.validate_auth(credentials: HTTPBasicCredentials | None)[source]

Validate HTTP Basic Authentication.

Authentication is disabled if KAVALAI_AGENT_BASIC_AUTH_USER and KAVALAI_AGENT_BASIC_AUTH_PASSWORD are not set in the environment.

Parameters:
credentials: HTTPBasicCredentials | None

The credentials provided in the request.

Returns:

True if authentication is successful or disabled.

Raises:

HTTPException – If authentication fails.

kavalai.server.session_scope(session_or_factory)[source]

Provide a database session from either a sessionmaker or an existing session.

This context manager ensures that if a factory is provided, a new session is created and closed properly. If an existing session is provided, it is used as-is.

async kavalai.server.handle_agent_run(engine: WorkflowEngine, session_provider: async_sessionmaker | None, input_data: dict, session_id: UUID | None = None, external_id: str | None = None)[source]

Execute the agent workflow with the provided input.

This is a standalone handler that can be used directly or wrapped in a FastAPI route.

Parameters:
engine: WorkflowEngine

The v2 WorkflowEngine to execute.

session_provider: async_sessionmaker | None

An optional SQLAlchemy async_sessionmaker for database sessions (unused, kept for compatibility).

input_data: dict

The input data for the workflow (already extracted from request).

session_id: UUID | None = None

Optional session ID for continuing a previous session.

external_id: str | None = None

Optional external identifier for tracking.

Returns:

A tuple of (session_id, output_data).

kavalai.server.format_sse_event(event: WorkflowStreamEvent) → str[source]

Render one WorkflowStreamEvent as an SSE frame.

async kavalai.server.stream_sse_events(events: AsyncIterable[WorkflowStreamEvent], ping_interval: float = 15.0, *, event_filter: Callable[[WorkflowStreamEvent], WorkflowStreamEvent | None] | None = None) → AsyncGenerator[str, None][source]

Format a workflow event stream as SSE frames with keepalive pings.

A : ping comment frame is emitted whenever no event arrives within ping_interval seconds (silent stretches such as long tool calls), so proxies don’t drop the connection. A WorkflowException from the engine ends the stream quietly — the engine has already emitted the workflow_failed event, and an SSE response cannot change its status code after the headers are sent.

event_filter sees every event before it is written, and returns the event to send, a modified copy, or None to drop it. See public_events().

The event stream is consumed by a single pump task (an async generator’s frames must run in one task — the engine binds task-scoped log context), while this generator races the queue against the ping timer. Closing this generator cancels the pump, which aborts the workflow run.

kavalai.server.sse_response(events: AsyncIterable[WorkflowStreamEvent], *, event_filter: Callable[[WorkflowStreamEvent], WorkflowStreamEvent | None] | None = None, headers: dict[str, str] | None = None, ping_interval: float = 15.0) → StreamingResponse[source]

Serve a workflow event stream as a Server-Sent Events response.

What the router’s /stream_agent returns, available to a host that serves a workflow from a route of its own — behind its own admission checks — so the frames, the keepalive and the headers stay the SDK’s. headers are added to SSE_HEADERS.

kavalai.server.create_default_auth_dependency() → Callable[source]

Create the default HTTP Basic Auth dependency using environment variables.

Returns:

A FastAPI dependency function that validates HTTP Basic Authentication.

kavalai.server.create_agent_router(engine: WorkflowEngine, session_provider: async_sessionmaker | None = None, auth_dependency: Callable | None = None, *, event_filter: Callable[[WorkflowStreamEvent], WorkflowStreamEvent | None] | None = None) → APIRouter[source]

Create a FastAPI router for a given workflow.

This function creates a reusable router that can be mounted on existing FastAPI applications, allowing for flexible composition and custom routing configurations.

The router serves POST /run_agent (blocking) and POST /stream_agent (SSE), both typed by the workflow’s own input and output data types, plus GET /workflow, GET /liveness and GET /health.

Parameters:
engine: WorkflowEngine

The WorkflowEngine instance to serve.

session_provider: async_sessionmaker | None = None

An optional SQLAlchemy async_sessionmaker to provide database sessions for agent execution.

auth_dependency: Callable | None = None

An optional FastAPI dependency for authentication. If None, the default HTTP Basic Auth will be used. Pass a custom dependency function to use your own auth, or pass lambda: None to disable authentication.

event_filter: Callable[[WorkflowStreamEvent], WorkflowStreamEvent | None] | None = None

Applied to every event /stream_agent sends. Pass public_events() when the stream is served to people other than the operator. /run_agent returns only the output, so it is unaffected.

Returns:

An APIRouter instance with configured endpoints.

Example

# Use with custom auth
def my_auth():
    # Custom auth logic
    pass

router = create_agent_router(engine, session_provider, auth_dependency=my_auth)
app.include_router(router, prefix="/agents/my-workflow")

# Or disable auth entirely
router = create_agent_router(engine, session_provider, auth_dependency=lambda: None)
kavalai.server.create_agent_app(engine: WorkflowEngine, session_provider: async_sessionmaker | None = None, auth_dependency: Callable | None = None, *, event_filter: Callable[[WorkflowStreamEvent], WorkflowStreamEvent | None] | None = None) → FastAPI[source]

Create a FastAPI application for a given workflow.

The application dynamically generates input and output models based on the workflow’s schema and provides endpoints to run the agent and retrieve its configuration.

Parameters:
engine: WorkflowEngine

The WorkflowEngine instance to serve.

session_provider: async_sessionmaker | None = None

An optional SQLAlchemy async_sessionmaker to provide database sessions for agent execution.

auth_dependency: Callable | None = None

An optional FastAPI dependency for authentication. If None, the default HTTP Basic Auth will be used.

event_filter: Callable[[WorkflowStreamEvent], WorkflowStreamEvent | None] | None = None

Applied to the streamed events; see create_agent_router().

Returns:

A FastAPI application instance.

kavalai.server.mask_db_uri(db_uri: str) → str[source]

Mask the password in a database URI.

Parameters:
db_uri: str

The database URI to mask.

Returns:

The masked database URI.

kavalai.server.load_provider_modules(module_names: str) → list[str][source]

Import modules that register custom LLM, embedding or RAG backends.

Each module is expected to call register_llm_provider(), register_embedding_provider() or register_rag_service() at import time. After they are loaded, every dotted registration is resolved, so a mistyped path fails here rather than on the first request that happens to reach that node.

Parameters:
module_names: str

Comma-separated dotted module names; blanks are ignored.

Returns:

The module names that were imported, in order.

Raises:

  • ImportError – A named module could not be imported.

  • RegistryError – A registration names a path that cannot be resolved.

kavalai.server.create_app_from_env_conf(workflow_path: str | None = None, db_uri: str | None = None, db_schema: str | None = None, pool_size: int | None = None, max_overflow: int | None = None, sql_echo: bool | None = None) → FastAPI[source]

Create Kavalai server application from environment configuration.

Optional parameters can override the environment variables.

The following environment variables are used:

  • KAVALAI_AGENT_WORKFLOW_PATH: Path to the workflow YAML file.

  • KAVALAI_AGENT_SETUP_MODULE: Optional module imported before the workflow is loaded, as a dotted name or a .py path. Registers the Python tools and named RAG services the workflow refers to.

  • KAVALAI_DB_URI: Database connection string.

  • KAVALAI_DB_SCHEMA: Database schema name.

  • KAVALAI_DB_POOL_SIZE: Database connection pool size (optional, default: 0).

  • KAVALAI_DB_MAX_OVERFLOW: Database connection pool max overflow (optional, default: 0).

  • KAVALAI_SQL_ECHO: Whether to log SQL queries (optional, default: False).

  • KAVALAI_PROVIDER_MODULES: Comma-separated modules to import before the workflow loads, so their backend registrations exist (optional).

  • KAVALAI_DEFAULT_LLM_MODEL: Model used when the workflow and its nodes name none, passed to the engine as default_llm_model (optional).

  • KAVALAI_LLM_TEMPERATURE, KAVALAI_LLM_TOP_P, KAVALAI_LLM_REASONING_EFFORT, KAVALAI_LLM_SERVICE_TIER, KAVALAI_LLM_TIMEOUT_SECONDS, KAVALAI_LLM_STREAM_TIMEOUT_SECONDS: fleet-wide defaults for every model call, passed to the engine as default_llm_parameters (optional).

  • KAVALAI_EMBEDDING_NORMALIZER_YAML: Normalizer installed as the default before the workflow loads (optional).

  • KAVALAI_RAG_MODEL: Embedding model of the default RAG service. Setting it registers that service without a setup module, over the index at KAVALAI_RAG_URI (required alongside it) in KAVALAI_RAG_SCHEMA (optional), with the normalizer above.

  • KAVALAI_AGENT_PUBLIC_EVENTS: Serve streams through public_events() (optional, default: false).

  • KAVALAI_AGENT_RUN_TIMEOUT_SECONDS: Seconds after which a run is cancelled and recorded as failed, passed to the engine as run_timeout (optional, default: no limit).

Parameters:
workflow_path: str | None = None

Path to the workflow YAML file.

db_uri: str | None = None

Database connection string.

db_schema: str | None = None

Database schema name.

pool_size: int | None = None

Database connection pool size.

max_overflow: int | None = None

Database connection pool max overflow.

sql_echo: bool | None = None

Whether to log SQL queries.

Returns:

A FastAPI application instance.

kavalai.server.run_agent_server()[source]

Start the Kaval.AI agent server using environment configuration.