API Server

Source Code: src/gaia/api/

Components: OpenAI-compatible API Server, Schemas, Endpoints, SSE Streaming Module: gaia.api Import: from gaia.api import app, schemas, agent_registry

Overview

The GAIA API Server provides an OpenAI-compatible REST API for exposing GAIA agents to external tools like VSCode, Claude Dev, and custom applications. It implements the OpenAI chat completions interface with streaming support, agent registration, and comprehensive debugging features. Key Features:

OpenAI-compatible endpoints (/v1/chat/completions, /v1/models)
Server-Sent Events (SSE) streaming
Dynamic agent registry
Workspace root extraction from GitHub Copilot
Debug modes (logging, prompts, step-through)
CORS support
Health checks
Token usage tracking

Requirements

Functional Requirements

Core Endpoints

POST /v1/chat/completions
- Non-streaming responses
- SSE streaming responses
- Message history: prior turns are loaded into the agent’s conversation, and system / developer messages are added to its system prompt
- temperature, top_p, and max_tokens apply to every model call the agent makes for the request
- Tool calls support (future)
GET /v1/models
- List available agents as models
- Model metadata (tokens, description)
GET /health
- Reports API, Lemonade LLM, and RAG component status
- Returns 200 OK while the API process is serving
- Uses top-level degraded when Lemonade is unavailable or no LLM is loaded

Agent Registry

Agent Registration
- Dynamic agent discovery
- Model ID mapping
- Agent instantiation
- Workspace configuration
Agent Management
- Singleton instances per agent type
- Silent mode operation
- Workspace root injection

Request Processing

Message Handling
- OpenAI message format; content is a string or an array of content parts
- Text parts are joined with newlines; any other part type (image_url, input_audio, file) is rejected with 400
- system and developer messages are appended to the agent’s system prompt
- The last message must be from the user (400 otherwise); earlier user/assistant/tool messages become the agent’s conversation history
Response Generation
- Agent process_query() integration
- Token counting
- Usage statistics
- Unique response IDs

Streaming Support

SSE Streaming
- Chunk-based streaming
- Delta content format
- Finish reason reporting
- Proper SSE formatting (data: prefix)
- A client disconnect stops the agent (its cancel event and the SSE handler’s cancelled flag are set), so it stops running tools and generating
- If the agent raises mid-stream, the stream ends with an {"error": {...}} chunk followed by data: [DONE]

Non-Functional Requirements

Performance
- Low latency response
- Efficient streaming
- Connection pooling
Reliability
- Error handling
- Connection recovery
- Graceful degradation
Debugging
- Request/response logging
- Prompt display
- Step-through mode
- Raw HTTP logging
Security
- CORS configuration
- Input validation
- Error sanitization
- Confirmation-gated tools (shell, file writes, MCP writes) are refused — the API has no approval channel. See SDK → API Server for the GAIA_AUTO_APPROVE_TOOLS opt-out.

API Specification

File Locations

src/gaia/api/app.py              # CLI entry point
src/gaia/api/openai_server.py    # FastAPI server
src/gaia/api/schemas.py          # Pydantic models
src/gaia/api/agent_registry.py   # Agent management
src/gaia/api/sse_handler.py      # SSE streaming

Schemas (Pydantic Models)

from typing import Any, Dict, List, Literal, Optional, Union
from pydantic import BaseModel, ConfigDict, Field

class ContentPart(BaseModel):
    """One element of a content-part array. Only type "text" is served."""
    model_config = ConfigDict(extra="allow")
    type: str
    text: Optional[str] = None

class ChatMessage(BaseModel):
    """Chat message in OpenAI format. "developer" is treated as "system"."""
    role: Literal["system", "developer", "user", "assistant", "tool"]
    content: Union[str, List[ContentPart], None] = None
    tool_calls: Optional[List[Dict[str, Any]]] = None
    tool_call_id: Optional[str] = None

class ChatCompletionRequest(BaseModel):
    """POST /v1/chat/completions request."""
    model: str = Field(..., description="Model ID — an id registered in AGENT_MODELS")
    messages: List[ChatMessage]
    stream: bool = Field(default=False, description="Enable SSE streaming")
    # Unset keeps the agent's own default. Set values apply to every model
    # call the agent makes while answering (max_tokens is per call).
    temperature: Optional[float] = Field(default=None, ge=0, le=2)
    max_tokens: Optional[int] = Field(default=None, gt=0)
    top_p: Optional[float] = Field(default=None, ge=0, le=1)

class ChatCompletionResponseMessage(BaseModel):
    """Response message from chat completion."""
    role: Literal["assistant"]
    content: str
    tool_calls: Optional[List[Dict[str, Any]]] = None

class ChatCompletionChoice(BaseModel):
    """A single completion choice."""
    index: int
    message: ChatCompletionResponseMessage
    finish_reason: Literal["stop", "length"]

class UsageInfo(BaseModel):
    """Token usage information."""
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

class ChatCompletionResponse(BaseModel):
    """POST /v1/chat/completions response (non-streaming)."""
    id: str
    object: Literal["chat.completion"]
    created: int
    model: str
    choices: List[ChatCompletionChoice]
    usage: UsageInfo

class ModelInfo(BaseModel):
    """Model metadata for /v1/models endpoint."""
    id: str
    object: Literal["model"]
    created: int
    owned_by: str
    description: Optional[str] = None
    max_input_tokens: Optional[int] = None
    max_output_tokens: Optional[int] = None

class ModelListResponse(BaseModel):
    """GET /v1/models response."""
    object: Literal["list"]
    data: List[ModelInfo]

API Endpoints

from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from gaia.api.local_http import build_caller_guard, cors_config

app = FastAPI(
    title="GAIA OpenAI-Compatible API",
    description="OpenAI-compatible API for GAIA agents",
    version="1.0.0",
)

# CORS: loopback origins only, plus GAIA_API_CORS_ORIGINS (gaia/api/local_http.py)
app.add_middleware(CORSMiddleware, **cors_config())

# Foreign browser origins -> 403; GAIA_API_KEY enforced when set
_chat_completions_guard = build_caller_guard(
    "POST /v1/chat/completions", public_paths=frozenset()
)

@app.post("/v1/chat/completions", dependencies=[Depends(_chat_completions_guard)])
async def create_chat_completion(request: ChatCompletionRequest):
    """
    Create chat completion (OpenAI-compatible endpoint).

    Supports both streaming (SSE) and non-streaming responses.

    Args:
        request: Chat completion request with model, messages, and options

    Returns:
        For non-streaming: ChatCompletionResponse
        For streaming: StreamingResponse with SSE chunks

    Raises:
        HTTPException 404: Model not found
        HTTPException 400: No user message in request

    Example Non-Streaming:
        POST /v1/chat/completions
        {
            "model": "your-agent-id",
            "messages": [{"role": "user", "content": "Write hello world"}],
            "stream": false
        }

        Response:
        {
            "id": "chatcmpl-abc123",
            "object": "chat.completion",
            "created": 1234567890,
            "model": "your-agent-id",
            "choices": [{
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": "Here's hello world:\\nprint('Hello, World!')"
                },
                "finish_reason": "stop"
            }],
            "usage": {
                "prompt_tokens": 10,
                "completion_tokens": 20,
                "total_tokens": 30
            }
        }

    Example Streaming:
        POST /v1/chat/completions
        {
            "model": "your-agent-id",
            "messages": [{"role": "user", "content": "Write hello world"}],
            "stream": true
        }

        Response (SSE):
        data: {"id":"chatcmpl-abc","object":"chat.completion.chunk",...}
        data: {"id":"chatcmpl-abc","object":"chat.completion.chunk",...}
        data: [DONE]
    """
    pass

@app.get("/v1/models")
async def list_models() -> ModelListResponse:
    """
    List available models (GAIA agents).

    Returns:
        ModelListResponse with list of available agents as models

    Example:
        GET /v1/models

        Response:
        {
            "object": "list",
            "data": [{"id": "gaia", "object": "model", ...}]
        }

        (AGENT_MODELS holds one entry, "gaia", the flagship agent. Register
        another in src/gaia/api/agent_registry.py to add to this list. An
        entry looks like:
        {
            "id": "your-agent-id",
            "object": "model",
            "created": 1234567890,
            "owned_by": "amd-gaia",
            "description": "...",
            "max_input_tokens": 32768,
            "max_output_tokens": 8192
        })
    """
    pass

@app.get("/health")
async def health_check():
    """
    Report API and backing-service health.

    Returns:
        Overall status and component-level status

    Example:
        GET /health

        Response:
        {
            "status": "ok",
            "service": "gaia-api",
            "components": {
                "api": {"status": "ready"},
                "llm": {
                    "status": "ready",
                    "backend": "lemonade",
                    "model": "Gemma-4-E4B-it-GGUF",
                    "url": "http://localhost:13305/api/v1"
                },
                "rag": {"status": "not_configured"}
            }
        }
    """
    llm = await _lemonade_health()
    return {
        "status": "ok" if llm["status"] == "ready" else "degraded",
        "service": "gaia-api",
        "components": {
            "api": {"status": "ready"},
            "llm": llm,
            "rag": {"status": "not_configured"},
        },
    }

Agent Registry

The API-server’s AgentRegistry is not a runtime plug-in registry — it is a hardcoded AGENT_MODELS dict in src/gaia/api/agent_registry.py plus a dynamic import helper. There is no register_agent() method today; to add a new model you edit AGENT_MODELS and restart the server.

# src/gaia/api/agent_registry.py (simplified)
from importlib import import_module
from typing import Any, Dict

# One entry ships: "gaia", the flagship agent. Add more here to expose them.
AGENT_MODELS: Dict[str, Dict[str, Any]] = {
    "gaia": {
        "class_name": "gaia_agent.agent.GaiaAgent",
        "init_params": {"silent_mode": True, "streaming": False},
    },
    # "your-agent-id": {
    #     "class_name": "my_package.my_agent.MyAPIAgent",
    #     "init_params": {"api_mode": True, "silent_mode": True},
    #     "description": "...",
    # },
}


class AgentRegistry:
    """Resolves model IDs to agent instances by importing the target class
    on demand. `gaia api start --debug/--show-prompts/--streaming/--step-through`
    set GAIA_API_* env-vars that are merged into `init_params` at startup."""

    def get_agent(self, model_id: str):
        """Load and instantiate the agent for `model_id`.

        Raises `KeyError` if the model is not registered.
        A *fresh* agent is created per call — there is no singleton cache.
        """
        entry = AGENT_MODELS[model_id]
        module_path, _, class_name = entry["class_name"].rpartition(".")
        agent_class = getattr(import_module(module_path), class_name)
        return agent_class(**entry["init_params"])

    def list_models(self):
        """Return the models block used by `/v1/models`."""
        return [
            {
                "id": model_id,
                "object": "model",
                "owned_by": "amd-gaia",
                "description": cfg.get("description", ""),
            }
            for model_id, cfg in AGENT_MODELS.items()
        ]

For the richer agent registry used by the Agent UI (not the OpenAI-style API server), see the Agent Registry spec covering src/gaia/agents/registry.py and ~/.gaia/agents/ discovery of Python custom agents.

### SSE Streaming

```python
from typing import AsyncGenerator
import json
import uuid
import time

async def stream_chat_completion(
    agent: Agent,
    user_message: str,
    model_id: str,
) -> AsyncGenerator[str, None]:
    """
    Stream chat completion using SSE format.

    Args:
        agent: Agent instance to use
        user_message: User's message
        model_id: Model identifier

    Yields:
        SSE-formatted chunks (data: {...})

    Example:
        async for chunk in stream_chat_completion(agent, "Hello", "your-agent-id"):
            print(chunk)
        # Output:
        # data: {"id":"chatcmpl-123","object":"chat.completion.chunk",...}
        # data: {"id":"chatcmpl-123","object":"chat.completion.chunk",...}
        # data: [DONE]
    """
    chunk_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
    created = int(time.time())

    try:
        # Stream response from agent
        for delta_content in agent.stream_query(user_message):
            chunk = {
                "id": chunk_id,
                "object": "chat.completion.chunk",
                "created": created,
                "model": model_id,
                "choices": [
                    {
                        "index": 0,
                        "delta": {"content": delta_content},
                        "finish_reason": None,
                    }
                ],
            }
            yield f"data: {json.dumps(chunk)}\n\n"

        # Final chunk with finish_reason
        final_chunk = {
            "id": chunk_id,
            "object": "chat.completion.chunk",
            "created": created,
            "model": model_id,
            "choices": [
                {
                    "index": 0,
                    "delta": {},
                    "finish_reason": "stop",
                }
            ],
        }
        yield f"data: {json.dumps(final_chunk)}\n\n"
        yield "data: [DONE]\n\n"

    except Exception as e:
        # Headers are already sent, so report the failure in-band.
        error_chunk = {
            "error": {
                "message": str(e),
                "type": "server_error",
            }
        }
        yield f"data: {json.dumps(error_chunk)}\n\n"
        yield "data: [DONE]\n\n"

Implementation Details

Request Processing Flow

async def create_chat_completion(request: ChatCompletionRequest):
    # 1. Validate the model FIRST: an unknown model is a 404 even when the
    #    message array is also malformed.
    if not registry.model_exists(request.model):
        raise HTTPException(status_code=404, detail=f"Model '{request.model}' not found")

    # 2. Split the request: the final user message is the query, system and
    #    developer messages are caller system text, the rest is history.
    #    Raises 400 on a non-text content part or a non-user last message.
    user_message, history, system_text = _split_conversation(request.messages)

    # 3. Resolve a fresh agent and load the conversation into it. get_agent
    #    raises ValueError for an unregistered model, re-raised as a 404.
    agent = registry.get_agent(request.model)
    agent.conversation_history = history
    if system_text:
        agent.set_caller_system_prompt(system_text)
    if history or system_text:
        agent.mark_external_content()  # caller-supplied, not typed this turn
    if request.temperature is not None:
        agent.chat.config.temperature = request.temperature
    # ...same for top_p and max_tokens

    # 4. Generate response
    if request.stream:
        # Streaming response
        return StreamingResponse(
            stream_chat_completion(agent, user_message, request.model),
            media_type="text/event-stream",
        )
    else:
        # Non-streaming response
        response_text = agent.process_query(user_message)

        # Token counting
        if isinstance(agent, ApiAgent):
            prompt_tokens = agent.estimate_tokens(user_message)
            completion_tokens = agent.estimate_tokens(response_text)
        else:
            prompt_tokens = len(user_message) // 4
            completion_tokens = len(response_text) // 4

        return ChatCompletionResponse(
            id=f"chatcmpl-{uuid.uuid4().hex[:8]}",
            object="chat.completion",
            created=int(time.time()),
            model=request.model,
            choices=[
                ChatCompletionChoice(
                    index=0,
                    message=ChatCompletionResponseMessage(
                        role="assistant",
                        content=response_text,
                    ),
                    finish_reason="stop",
                )
            ],
            usage=UsageInfo(
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                total_tokens=prompt_tokens + completion_tokens,
            ),
        )

Debug Middleware

@app.middleware("http")
async def log_raw_requests(request: Request, call_next):
    """
    Middleware to log raw HTTP requests when debug mode is enabled.
    """
    if os.environ.get("GAIA_API_DEBUG") == "1":
        logger.debug("=" * 80)
        logger.debug("RAW HTTP REQUEST")
        logger.debug("=" * 80)
        logger.debug(f"Path: {request.url.path}")
        logger.debug(f"Method: {request.method}")
        logger.debug("Headers:")
        for name, value in request.headers.items():
            logger.debug(f"  {name}: {value}")

        # Don't read body for streaming endpoints
        if request.url.path != "/v1/chat/completions":
            body_bytes = await request.body()
            logger.debug(f"Body: {body_bytes.decode('utf-8')}")

        logger.debug("=" * 80)

    response = await call_next(request)
    return response

Testing Requirements

Unit Tests

File: tests/api/test_api_server.py

import pytest
from fastapi.testclient import TestClient
from gaia.api.openai_server import app

client = TestClient(app)

def test_health_check():
    """Test health endpoint."""
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] in ("ok", "degraded")
    assert data["service"] == "gaia-api"
    assert data["components"]["api"] == {"status": "ready"}

def test_list_models():
    """Test models listing."""
    response = client.get("/v1/models")
    assert response.status_code == 200
    data = response.json()
    assert data["object"] == "list"
    assert isinstance(data["data"], list)
    assert len(data["data"]) > 0

def test_chat_completion_non_streaming():
    """Test non-streaming chat completion."""
    response = client.post(
        "/v1/chat/completions",
        json={
            "model": "your-agent-id",
            "messages": [{"role": "user", "content": "Say hello"}],
            "stream": False
        }
    )
    assert response.status_code == 200
    data = response.json()
    assert data["object"] == "chat.completion"
    assert len(data["choices"]) > 0
    assert data["choices"][0]["message"]["role"] == "assistant"
    assert isinstance(data["choices"][0]["message"]["content"], str)
    assert "usage" in data

def test_chat_completion_streaming():
    """Test streaming chat completion."""
    with client.stream(
        "POST",
        "/v1/chat/completions",
        json={
            "model": "your-agent-id",
            "messages": [{"role": "user", "content": "Count to 3"}],
            "stream": True
        }
    ) as response:
        assert response.status_code == 200
        assert response.headers["content-type"] == "text/event-stream"

        chunks = []
        for line in response.iter_lines():
            if line.startswith("data: "):
                data = line[6:]  # Remove "data: " prefix
                if data != "[DONE]":
                    import json
                    chunk = json.loads(data)
                    chunks.append(chunk)

        assert len(chunks) > 0
        assert chunks[0]["object"] == "chat.completion.chunk"

def test_chat_completion_invalid_model():
    """Test chat completion with invalid model."""
    response = client.post(
        "/v1/chat/completions",
        json={
            "model": "nonexistent-model",
            "messages": [{"role": "user", "content": "Hello"}],
        }
    )
    assert response.status_code == 404

def test_chat_completion_no_user_message():
    """Test chat completion without user message."""
    response = client.post(
        "/v1/chat/completions",
        json={
            "model": "your-agent-id",
            "messages": [{"role": "system", "content": "You are helpful"}],
        }
    )
    assert response.status_code == 400

Usage Examples

Example 1: Start API Server

# Foreground
gaia api start

# Background
gaia api start --background

# Custom host/port (binding beyond loopback requires GAIA_API_KEY)
GAIA_API_KEY=$(openssl rand -hex 32) gaia api start --host 0.0.0.0 --port 9090

# With debug
gaia api start --debug --show-prompts

# Stop server
gaia api stop

Example 2: Non-Streaming Request (curl)

curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "your-agent-id",
    "messages": [
      {"role": "user", "content": "Write a hello world program"}
    ],
    "stream": false
  }'

Example 3: Streaming Request (curl)

curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "your-agent-id",
    "messages": [
      {"role": "user", "content": "Write a hello world program"}
    ],
    "stream": true
  }'

Example 4: Python Client

import requests

# List models
response = requests.get("http://localhost:8080/v1/models")
models = response.json()["data"]
for model in models:
    print(f"{model['id']}: {model.get('description', 'No description')}")

# Non-streaming chat
response = requests.post(
    "http://localhost:8080/v1/chat/completions",
    json={
        "model": "your-agent-id",
        "messages": [
            {"role": "user", "content": "Write hello world in Python"}
        ],
        "stream": False
    }
)
result = response.json()
print(result["choices"][0]["message"]["content"])

# Streaming chat
response = requests.post(
    "http://localhost:8080/v1/chat/completions",
    json={
        "model": "your-agent-id",
        "messages": [
            {"role": "user", "content": "Count to 5"}
        ],
        "stream": True
    },
    stream=True
)

for line in response.iter_lines():
    if line:
        line = line.decode('utf-8')
        if line.startswith("data: "):
            data = line[6:]
            if data != "[DONE]":
                import json
                chunk = json.loads(data)
                content = chunk["choices"][0]["delta"].get("content", "")
                if content:
                    print(content, end="", flush=True)
print()

Example 5: OpenAI Python SDK

from openai import OpenAI

# Point to GAIA API server
client = OpenAI(
    base_url="http://localhost:8080/v1",
    api_key="none"  # Not required for GAIA
)

# Non-streaming
response = client.chat.completions.create(
    model="your-agent-id",
    messages=[
        {"role": "user", "content": "Write hello world"}
    ]
)
print(response.choices[0].message.content)

# Streaming
stream = client.chat.completions.create(
    model="your-agent-id",
    messages=[
        {"role": "user", "content": "Count to 5"}
    ],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

CLI Interface

Commands

# app.py

def start_server(
    host: str = "localhost",
    port: int = 8080,
    background: bool = False,
    debug: bool = False,
    show_prompts: bool = False,
    streaming: bool = False,
    step_through: bool = False,
):
    """
    Start the API server.

    Args:
        host: Host to bind to (default: localhost)
        port: Port to bind to (default: 8080)
        background: Run in background if True
        debug: Enable debug logging
        show_prompts: Display prompts sent to LLM
        streaming: Enable real-time streaming of LLM responses
        step_through: Enable step-through debugging mode
    """
    pass

def stop_server(port: int = 8080):
    """
    Stop the API server by finding and killing processes on the port.

    Args:
        port: Port number to stop server on (default: 8080)
    """
    pass

def check_status():
    """Check if API server is running."""
    pass

API Server Technical Specification

VS Code client authentication

Run GAIA: Set API Key from the VS Code command palette and enter the same value as the server’s GAIA_API_KEY. The extension stores it in VS Code SecretStorage and uses it for model discovery and chat completions. Leave the input blank to clear it; cancelling preserves the existing key. Keyless loopback servers continue to work.

Core Framework

SDKs

Infrastructure

Code Infrastructure

Tool Mixins

Design

Packaging

Agents & Apps

Overview

Requirements

Functional Requirements

Core Endpoints

Agent Registry

Request Processing

Streaming Support

Non-Functional Requirements

API Specification

File Locations

Schemas (Pydantic Models)

API Endpoints

Agent Registry

Implementation Details

Request Processing Flow

Debug Middleware

Testing Requirements

Unit Tests

Usage Examples

Example 1: Start API Server

Example 2: Non-Streaming Request (curl)

Example 3: Streaming Request (curl)

Example 4: Python Client

Example 5: OpenAI Python SDK

CLI Interface

Commands

VS Code client authentication

​Overview

​Requirements

​Functional Requirements

​Core Endpoints

​Agent Registry

​Request Processing

​Streaming Support

​Non-Functional Requirements

​API Specification

​File Locations

​Schemas (Pydantic Models)

​API Endpoints

​Agent Registry

​Implementation Details

​Request Processing Flow

​Debug Middleware

​Testing Requirements

​Unit Tests

​Usage Examples

​Example 1: Start API Server

​Example 2: Non-Streaming Request (curl)

​Example 3: Streaming Request (curl)

​Example 4: Python Client

​Example 5: OpenAI Python SDK

​CLI Interface

​Commands

​VS Code client authentication

Overview

Requirements

Functional Requirements

Core Endpoints

Agent Registry

Request Processing

Streaming Support

Non-Functional Requirements

API Specification

File Locations

Schemas (Pydantic Models)

API Endpoints

Agent Registry

Implementation Details

Request Processing Flow

Debug Middleware

Testing Requirements

Unit Tests

Usage Examples

Example 1: Start API Server

Example 2: Non-Streaming Request (curl)

Example 3: Streaming Request (curl)

Example 4: Python Client

Example 5: OpenAI Python SDK

CLI Interface

Commands

VS Code client authentication