calute.api_server.server#

Main API server for the modular Calute API server.

This module provides the core API server infrastructure for Calute, including: - FastAPI-based HTTP server with OpenAI-compatible endpoints - Agent registration and management - Cortex multi-agent orchestration support - Modular router architecture for different endpoint groups - Completion services for both standard and Cortex agents

The server supports both standard Calute agents and Cortex agents for multi-agent orchestration, with full compatibility with OpenAI client libraries.

class calute.api_server.server.CaluteAPIServer(calute_instance: calute.calute.Calute | None = None, llm: calute.llms.base.BaseLLM | None = None, can_overide_samplings: bool = False, enable_cortex: bool = False, use_universal_agent: bool = True)[source]#

Bases: object

Modular FastAPI server that provides OpenAI-compatible API for Calute agents.

This server exposes registered Calute agents through HTTP endpoints that follow the OpenAI API specification, allowing seamless integration with OpenAI client libraries.

The server is designed with a modular architecture: - Separate routers for different endpoint groups - Dedicated service for completion logic - Message conversion utilities - Centralized models for request/response handling

calute#

The Calute instance managing agents.

llm#

LLM instance for Cortex agents.

agents#

Dictionary mapping agent IDs to Agent objects.

cortex_agents#

List of registered CortexAgent instances.

enable_cortex#

Whether Cortex endpoints are enabled.

app#

FastAPI application instance.

completion_service#

Service for handling standard chat completions.

cortex_completion_service#

Service for handling Cortex completions.

Example

>>> from calute import Calute
>>> from calute.api_server import CaluteAPIServer
>>>
>>> calute = Calute(client=openai_client)
>>> server = CaluteAPIServer(calute)
>>> server.register_agent(my_agent)
>>> server.run(port=8000)
classmethod create_server(client: Any, agents: list[calute.types.agent_types.Agent] | None | calute.types.agent_types.Agent = None, can_overide_samplings: bool = False, **calute_kwargs) CaluteAPIServer[source]#

Create a Calute API server with the given client and agents.

This is a convenience factory method that handles the full setup sequence: creating a Calute instance, wrapping it in a CaluteAPIServer, and registering all provided agents. The returned server is ready to be started with run().

Parameters
  • client – An OpenAI-compatible client instance (e.g., openai.OpenAI(...)). Passed to the Calute constructor.

  • agents – A single Agent instance or a list of Agent instances to register with the server. If None, no agents are registered and they must be added later via register_agent().

  • can_overide_samplings – Whether to allow incoming request parameters (temperature, top_p, max_tokens, etc.) to override the agent’s default sampling settings. Defaults to False.

  • **calute_kwargs – Additional keyword arguments passed directly to the Calute constructor (e.g., max_history_length, system_prompt).

Returns

A fully configured CaluteAPIServer instance with all provided agents registered and ready to serve requests.

Example

>>> import openai
>>> from calute.types import Agent
>>> from calute.api_server import CaluteAPIServer
>>>
>>> client = openai.OpenAI(api_key="key", base_url="url")
>>> agent = Agent(id="assistant", model="gpt-4", instructions="Help users")
>>> server = CaluteAPIServer.create_server(client, agents=[agent])
>>> server.run(port=8000)
register_agent(agent: Agent) None[source]#

Register a standard agent to be available via the API.

Adds the agent to both the Calute instance and the server’s internal agent registry. The agent becomes accessible through the chat completions endpoint using its ID, name, or model as the model parameter in requests. If routers have not yet been initialized, this method triggers router setup.

Parameters

agent – The Agent instance to register. Must have at least one of id, name, or model set to serve as the lookup key in the agent registry.

Raises

ValueError – If no Calute instance was provided during server initialization, since standard agents require Calute for execution.

Example

>>> server = CaluteAPIServer(calute_instance=calute)
>>> agent = Agent(id="assistant", model="gpt-4", instructions="Help users")
>>> server.register_agent(agent)
register_cortex_agent(agent: CortexAgent) None[source]#

Register a CortexAgent for multi-agent orchestration.

Adds the agent to the Cortex agent pool. If the Cortex completion service has already been initialized, its agent list is updated immediately. If routers have not yet been initialized, this method triggers router setup.

Parameters

agent – The CortexAgent instance to register. This agent will be available for task assignment and orchestration through the Cortex completion service.

Raises

ValueError – If Cortex was not enabled during server initialization (i.e., enable_cortex=False).

Example

>>> server = CaluteAPIServer(llm=my_llm, enable_cortex=True)
>>> cortex_agent = CortexAgent(name="researcher", llm=my_llm)
>>> server.register_cortex_agent(cortex_agent)
run(host: str = '0.0.0.0', port: int = 11881, **kwargs) None[source]#

Run the API server using uvicorn.

Starts the uvicorn ASGI server with the configured FastAPI application. If routers have not been initialized yet, this method attempts to set them up. Raises an error if no agents have been registered and Cortex is not enabled.

Parameters
  • host – The hostname or IP address to bind the server to. Defaults to "0.0.0.0" (all interfaces).

  • port – The TCP port number to bind the server to. Defaults to 11881.

  • **kwargs – Additional keyword arguments passed directly to uvicorn.run(), such as log_level, workers, ssl_keyfile, etc.

Raises

RuntimeError – If no agents are registered and Cortex is not enabled, since the server would have no endpoints to serve.

Example

>>> server = CaluteAPIServer(calute_instance=calute)
>>> server.register_agent(agent)
>>> server.run(host="127.0.0.1", port=8000, log_level="info")