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:
objectModular 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
Caluteinstance, wrapping it in aCaluteAPIServer, and registering all provided agents. The returned server is ready to be started withrun().- Parameters
client – An OpenAI-compatible client instance (e.g.,
openai.OpenAI(...)). Passed to theCaluteconstructor.agents – A single
Agentinstance or a list ofAgentinstances to register with the server. IfNone, no agents are registered and they must be added later viaregister_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
Caluteconstructor (e.g.,max_history_length,system_prompt).
- Returns
A fully configured
CaluteAPIServerinstance 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
modelparameter in requests. If routers have not yet been initialized, this method triggers router setup.- Parameters
agent – The
Agentinstance to register. Must have at least one ofid,name, ormodelset to serve as the lookup key in the agent registry.- Raises
ValueError – If no
Caluteinstance 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
CortexAgentfor 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
CortexAgentinstance 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 aslog_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")