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 kavalai.client import AgentClient

client = AgentClient("http://localhost:8000")
async for chunk in client.stream_agent(Message(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.

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.

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) → 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.

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.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) → 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.

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) → 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.

This function now uses create_agent_router() internally for better composability.

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.

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.

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_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_OPENAI_SERVICE_TIER: The service tier to use for OpenAI API calls (optional, e.g. “priority”).

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.
openai_service_tier: str | None = None¶: The service tier to use for OpenAI API calls.

Returns:¶

A FastAPI application instance.

kavalai.server.run_agent_server()[source]¶: Start the Kaval.AI agent server using environment configuration.