Skip to main content

routes.agents

Agent management routes for dynamic agent creation and registration.

Provides REST API endpoints for:

  • Creating new agents
  • Listing agents
  • Getting agent details
  • Deleting agents

get_stored_agent_spec​

def get_stored_agent_spec(agent_id: str) -> dict[str, Any] | None

Get the original creation spec for an agent.

set_api_prefix​

def set_api_prefix(prefix: str) -> None

Set the API prefix for dynamic mount paths.

get_agent_spec_library​

@router.get("/library", response_model=list[AgentSpec])
async def get_agent_spec_library() -> list[dict[str, Any]]

Get all available agent specifications from the library.

Returns predefined agent templates that can be used to create new agents.

get_agent_spec​

@router.get("/library/{agent_id:path}", response_model=AgentSpec)
async def get_agent_spec(agent_id: str) -> dict[str, Any]

Get a specific agent specification from the library.

Arguments:

  • agent_id - The ID of the agent spec (e.g., 'data-acquisition', 'crawler')

McpServerSelection Objects​

class McpServerSelection(BaseModel)

Selection of an MCP server with its origin.

Attributes:

  • id - Unique identifier of the MCP server (McpId)
  • origin - Origin of the server - 'config' (from mcp.json) or 'catalog' (built-in)

CreateAgentRequest Objects​

class CreateAgentRequest(BaseModel)

Request body for creating a new agent.

CreateAgentResponse Objects​

class CreateAgentResponse(BaseModel)

Response after creating an agent.

AgentListResponse Objects​

class AgentListResponse(BaseModel)

Response for listing agents.

create_agent​

@router.post("", response_model=CreateAgentResponse)
async def create_agent(request: CreateAgentRequest,
                       http_request: Request) -> CreateAgentResponse

Create and register a new agent.

This endpoint dynamically creates an agent based on the specified configuration and registers it with the appropriate transport protocols.

Arguments:

  • request - Agent creation parameters.
  • http_request - The HTTP request object (to access the FastAPI app).

Returns:

Information about the created agent.

Raises:

  • HTTPException - If agent creation fails.

list_agents​

@router.get("", response_model=AgentListResponse)
async def list_agents() -> AgentListResponse

List all registered agents.

Returns:

List of agent information including toolset details.

get_agent​

@router.get("/{agent_id:path}")
async def get_agent(agent_id: str) -> dict[str, Any]

Get information about a specific agent.

Arguments:

  • agent_id - The agent identifier.

Returns:

Agent information with full details.

Raises:

  • HTTPException - If agent not found.

delete_agent​

@router.delete("/{agent_id:path}")
async def delete_agent(agent_id: str) -> dict[str, str]

Delete an agent.

Arguments:

  • agent_id - The agent identifier.

Returns:

Success message.

Raises:

  • HTTPException - If agent not found.

UpdateAgentTransportRequest Objects​

class UpdateAgentTransportRequest(BaseModel)

Request to update an agent's transport protocol.

update_agent_transport​

@router.patch("/{agent_id:path}/transport")
async def update_agent_transport(agent_id: str,
                                 request: UpdateAgentTransportRequest,
                                 http_request: Request) -> dict[str, Any]

Update an agent's transport protocol (ag-ui or vercel-ai).

This re-registers the agent with the new transport while keeping the same underlying agent instance.

Arguments:

  • agent_id - The agent identifier.
  • request - The transport update request.
  • http_request - The FastAPI request.

Returns:

Updated agent information.

UpdateAgentMcpServersRequest Objects​

class UpdateAgentMcpServersRequest(BaseModel)

Request to update an agent's MCP servers.

update_agent_mcp_servers​

@router.patch("/{agent_id:path}/mcp-servers")
async def update_agent_mcp_servers(
        agent_id: str,
        request: UpdateAgentMcpServersRequest) -> dict[str, Any]

Update an agent's selected MCP servers at runtime.

This allows dynamically adding or removing MCP servers from a running agent without recreating the agent.

Arguments:

  • agent_id - The agent identifier.
  • request - The new list of MCP server IDs.

Returns:

Updated agent info.

Raises:

  • HTTPException - If agent not found or update fails.

ConfigureSandboxRequest Objects​

class ConfigureSandboxRequest(BaseModel)

Request to configure the code sandbox manager.

SandboxStatusResponse Objects​

class SandboxStatusResponse(BaseModel)

Response with current sandbox status.

get_sandbox_status​

@router.get("/sandbox/status")
async def get_sandbox_status() -> SandboxStatusResponse

Get the current status of the code sandbox manager.

Returns:

Current sandbox configuration and status.

configure_sandbox​

@router.post("/sandbox/configure")
async def configure_sandbox(
        request: ConfigureSandboxRequest) -> SandboxStatusResponse

Configure the code sandbox manager.

This endpoint allows runtime configuration of the sandbox variant. Use 'local-eval' for simple Python exec-based execution, 'jupyter' for a managed Jupyter sandbox, or 'local-jupyter' to connect to an existing Jupyter server.

Note: If a sandbox is currently running with a different configuration, it will be stopped and recreated on next use.

Arguments:

  • request - Sandbox configuration including variant and Jupyter details.

Returns:

Updated sandbox status.

Raises:

  • HTTPException - If configuration fails.

restart_sandbox​

@router.post("/sandbox/restart")
async def restart_sandbox() -> SandboxStatusResponse

Restart the code sandbox with current configuration.

This stops any running sandbox and creates a new instance with the current configuration.

Returns:

Updated sandbox status.

EnvVar Objects​

class EnvVar(BaseModel)

Environment variable name-value pair.

StartAgentMcpServersRequest Objects​

class StartAgentMcpServersRequest(BaseModel)

Request to start MCP servers for a running agent.

For two-container Kubernetes deployments, both jupyter_sandbox and mcp_proxy_url should be provided to enable the full codemode flow:

  • jupyter_sandbox: URL of the Jupyter server for code execution
  • mcp_proxy_url: URL of the MCP proxy endpoint for tool calls

In Kubernetes pods, containers communicate via localhost (127.0.0.1).

AgentMcpServersResponse Objects​

class AgentMcpServersResponse(BaseModel)

Response for agent MCP server operations.

start_all_agents_mcp_servers​

@router.post("/mcp-servers/start")
async def start_all_agents_mcp_servers(
        body: StartAgentMcpServersRequest,
        request: Request) -> AgentMcpServersResponse

Start catalog MCP servers for all running agents.

This endpoint starts the MCP servers that are configured in each agent's selected_mcp_servers list. Environment variables can be provided to configure the servers (e.g., API keys).

If an agent has Codemode enabled, the Codemode toolset will be rebuilt to include the newly started servers as programmatic tools.

If jupyter_sandbox is provided, the code sandbox manager will be configured to use the Jupyter kernel for code execution instead of local eval.

Arguments:

  • body - Environment variables and optional jupyter_sandbox URL.
  • request - FastAPI request object (provides access to app.state).

Returns:

Aggregated status of server start operations across all agents.

start_agent_mcp_servers​

@router.post("/{agent_id:path}/mcp-servers/start")
async def start_agent_mcp_servers(agent_id: str,
                                  body: StartAgentMcpServersRequest,
                                  request: Request) -> AgentMcpServersResponse

Start catalog MCP servers defined for a specific running agent.

This endpoint starts the MCP servers that are configured in the agent's selected_mcp_servers list. Environment variables can be provided to configure the servers (e.g., API keys).

If the agent has Codemode enabled, the Codemode toolset will be rebuilt to include the newly started servers as programmatic tools.

If jupyter_sandbox is provided, the code sandbox manager will be configured to use the Jupyter kernel for code execution instead of local eval.

Arguments:

  • agent_id - The agent identifier.
  • body - Environment variables and optional jupyter_sandbox URL.
  • request - FastAPI request object (provides access to app.state).

Returns:

Status of each server start operation.

Raises:

  • HTTPException - If agent not found or operation fails.

stop_all_agents_mcp_servers​

@router.post("/mcp-servers/stop")
async def stop_all_agents_mcp_servers() -> AgentMcpServersResponse

Stop catalog MCP servers for all running agents.

This endpoint stops the MCP servers that are configured in each agent's selected_mcp_servers list.

Returns:

Aggregated status of server stop operations across all agents.

stop_agent_mcp_servers​

@router.post("/{agent_id:path}/mcp-servers/stop")
async def stop_agent_mcp_servers(agent_id: str) -> AgentMcpServersResponse

Stop catalog MCP servers for a specific running agent.

This endpoint stops the MCP servers that are configured in the agent's selected_mcp_servers list.

Arguments:

  • agent_id - The agent identifier.

Returns:

Status of each server stop operation.

Raises:

  • HTTPException - If agent not found or operation fails.

prepare_checkpoint_endpoint​

@router.post("/prepare-checkpoint")
async def prepare_checkpoint_endpoint(http_request: Request) -> dict[str, Any]

Prepare the agent-runtimes service for a CRIU checkpoint.

Flushes DBOS state and closes connections to allow clean container freeze. Called by the companion before criu dump.

post_restore_endpoint​

@router.post("/post-restore")
async def post_restore_endpoint(http_request: Request) -> dict[str, Any]

Re-initialize after a CRIU restore.

Re-launches DBOS and re-establishes connections after container thaw. Called by the companion after criu restore.

ConfigureFromSpecRequest Objects​

class ConfigureFromSpecRequest(BaseModel)

Request body for the configure-from-spec endpoint.

configure_from_spec_endpoint​

@router.post("/configure-from-spec")
async def configure_from_spec_endpoint(
        http_request: Request,
        body: ConfigureFromSpecRequest) -> dict[str, Any]

Configure the running default agent from an AgentSpec ID.

Called by the companion during run-start-hooks when the operator provides an agent_spec_id.

The stored agentspec is kept in memory. When an update is triggered, the incoming spec is compared with the stored one. If they differ the PydanticAI agent is deleted and recreated with the new spec. The running sandbox is not restarted — CodeSandboxManager is a singleton whose lifecycle is independent of the agent.

When jupyter_sandbox / mcp_proxy_url are provided the sandbox is configured in the same request, removing the need for a separate mcp-servers/start call from the companion.

TriggerRunRequest Objects​

class TriggerRunRequest(BaseModel)

Body for POST /{agent_id}/trigger/run.

trigger_run​

@router.post("/{agent_id}/trigger/run")
async def trigger_run(agent_id: str, body: TriggerRunRequest,
                      request: Request) -> dict[str, Any]

Manually trigger an agent run (e.g. the once trigger).

The endpoint looks up the agent, retrieves its spec trigger config, creates the appropriate invoker, and fires it in the background.