API Server Usage Guide¶

This guide explains how to use the aipartnerupflow API server for remote task execution and integration.

Overview¶

The API server provides: - A2A Protocol Server: Standard agent-to-agent communication protocol (default) - MCP Server: Model Context Protocol server exposing task orchestration as MCP tools and resources - HTTP API: RESTful endpoints for task management - Real-time Streaming: Progress updates via SSE/WebSocket - Multi-user Support: User isolation and authentication

Starting the API Server¶

Basic Startup¶

# Start server on default port (8000) with A2A Protocol (default)
aipartnerupflow serve

# Or use the server command
aipartnerupflow-server

# Or use Python module
python -m aipartnerupflow.api.main

Protocol Selection¶

You can choose which protocol to use via the AIPARTNERUPFLOW_API_PROTOCOL environment variable:

# A2A Protocol Server (default)
export AIPARTNERUPFLOW_API_PROTOCOL=a2a
python -m aipartnerupflow.api.main

# MCP Server
export AIPARTNERUPFLOW_API_PROTOCOL=mcp
python -m aipartnerupflow.api.main

Supported Protocols: - a2a (default): A2A Protocol Server for agent-to-agent communication - mcp: MCP (Model Context Protocol) Server exposing task orchestration as MCP tools and resources

Advanced Options¶

# Custom host and port
aipartnerupflow serve --host 0.0.0.0 --port 8080

# Enable auto-reload (development)
aipartnerupflow serve --reload

# Multiple workers (production)
aipartnerupflow serve --workers 4

# Custom configuration
aipartnerupflow serve --config config.yaml

API Endpoints¶

Protocol Selection¶

The API server supports multiple protocols:

1. A2A Protocol (default): Standard agent-to-agent communication protocol
2. MCP Protocol: Model Context Protocol for tool and resource access

A2A Protocol Endpoints¶

The API server implements the A2A (Agent-to-Agent) Protocol standard when AIPARTNERUPFLOW_API_PROTOCOL=a2a (default).

Get Agent Card¶

curl http://localhost:8000/.well-known/agent-card

Returns agent capabilities and available skills.

Execute Task Tree (A2A Protocol)¶

curl -X POST http://localhost:8000/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks.execute",
    "params": {
      "tasks": [
        {
          "id": "task1",
          "name": "my_task",
          "user_id": "user123",
          "inputs": {"key": "value"}
        }
      ]
    },
    "id": "request-123"
  }'

Note: The method execute_task_tree is still supported for backward compatibility, but tasks.execute is the recommended standard method name.

Task Management via A2A Protocol¶

All task management operations are now fully supported through the A2A Protocol / route:

• Task Execution: tasks.execute (or execute_task_tree for backward compatibility)
• Task CRUD: tasks.create, tasks.get, tasks.update, tasks.delete
• Task Query: tasks.detail, tasks.tree, tasks.list, tasks.children
• Running Tasks: tasks.running.list, tasks.running.status, tasks.running.count
• Task Control: tasks.cancel, tasks.copy
• Task Generation: tasks.generate (generate task tree from natural language using LLM)

All methods follow the same A2A Protocol JSON-RPC format and return A2A Protocol Task objects with real-time status updates.

Task Management Endpoints (Legacy JSON-RPC)¶

Create Tasks¶

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks.create",
    "params": {
      "tasks": [...]
    },
    "id": "request-123"
  }'

Get Task Status¶

curl http://localhost:8000/tasks/{task_id}/status

List Tasks¶

curl http://localhost:8000/tasks?user_id=user123

Streaming Support¶

Server-Sent Events (SSE)¶

Use tasks.execute with use_streaming=true to receive real-time updates via SSE:

curl -N -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "method": "tasks.execute", "params": {"task_id": "task-123", "use_streaming": true}, "id": 1}' \
  http://localhost:8000/tasks

The response will be a Server-Sent Events stream with real-time progress updates.

Copy Before Execution¶

Use tasks.execute with copy_execution=true to copy the task before execution, preserving the original task's execution history:

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "method": "tasks.execute", "params": {"task_id": "task-123", "copy_execution": true}, "id": 1}'

To also copy child tasks with their dependencies:

curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc": "2.0", "method": "tasks.execute", "params": {"task_id": "task-123", "copy_execution": true, "copy_children": true}, "id": 1}'

The response will include both task_id (the copied task) and original_task_id (the original task).

WebSocket¶

Connect via WebSocket for bidirectional communication:

const ws = new WebSocket('ws://localhost:8000/ws');
ws.onmessage = (event) => {
  const update = JSON.parse(event.data);
  console.log('Progress:', update.progress);
};

Client Integration¶

Python Client Example¶

import httpx
import json

# Execute task via API
async with httpx.AsyncClient() as client:
    response = await client.post(
        "http://localhost:8000/",
        json={
            "jsonrpc": "2.0",
            "method": "tasks.execute",
            "params": {
                "tasks": [
                    {
                        "id": "task1",
                        "name": "my_task",
                        "user_id": "user123",
                        "inputs": {"key": "value"}
                    }
                ]
            },
            "id": "request-123"
        }
    )
    result = response.json()
    print(result)

JavaScript Client Example¶

// Execute task via API
const response = await fetch('http://localhost:8000/', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    jsonrpc: '2.0',
    method: 'tasks.execute',
    params: {
      tasks: [
        {
          id: 'task1',
          name: 'my_task',
          user_id: 'user123',
          inputs: { key: 'value' }
        }
      ]
    },
    id: 'request-123'
  })
});

const result = await response.json();
console.log(result);

Authentication¶

JWT Authentication (Optional)¶

If authentication is enabled:

# Get token
curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username": "user", "password": "pass"}'

# Use token
curl -X POST http://localhost:8000/ \
  -H "Authorization: Bearer {token}" \
  -H "Content-Type: application/json" \
  -d '{...}'

LLM API Key Management¶

The API server supports dynamic LLM API key injection for CrewAI tasks. Keys can be provided via request headers or user configuration.

Request Header (Demo/One-time Usage)¶

For demo or one-time usage, you can provide LLM API keys via the X-LLM-API-KEY header:

# Simple format (auto-detects provider from model)
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "X-LLM-API-KEY: sk-your-openai-key" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks.create",
    "params": {
      "tasks": [{
        "id": "task1",
        "name": "CrewAI Task",
        "schemas": {"method": "crewai_executor"},
        "params": {
          "works": {
            "agents": {
              "researcher": {
                "role": "Research Analyst",
                "llm": "openai/gpt-4"
              }
            }
          }
        }
      }]
    }
  }'

# Provider-specific format (explicit provider)
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "X-LLM-API-KEY: openai:sk-your-openai-key" \
  -d '{...}'

# Anthropic example
curl -X POST http://localhost:8000/tasks \
  -H "Content-Type: application/json" \
  -H "X-LLM-API-KEY: anthropic:sk-ant-your-key" \
  -d '{...}'

Header Format: - Simple: X-LLM-API-KEY: <api-key> (provider auto-detected from model name) - Provider-specific: X-LLM-API-KEY: <provider>:<api-key> (e.g., openai:sk-xxx, anthropic:sk-ant-xxx)

Supported Providers: - openai - OpenAI (GPT models) - anthropic - Anthropic (Claude models) - google / gemini - Google (Gemini models) - mistral - Mistral AI - groq - Groq - And more (see LLM Key Injector documentation)

Priority: 1. Request header (X-LLM-API-KEY) - highest priority 2. User config (if llm-key-config extension is installed) 3. Environment variables (automatically read by CrewAI/LiteLLM)

User Configuration (Multi-user Scenarios)¶

For production multi-user scenarios, use the llm-key-config extension:

# Install extension
pip install aipartnerupflow[llm-key-config]

Then configure keys programmatically:

from aipartnerupflow.extensions.llm_key_config import LLMKeyConfigManager

# Set user's LLM key
config_manager = LLMKeyConfigManager()
config_manager.set_key(user_id="user123", api_key="sk-xxx", provider="openai")

# Set provider-specific keys
config_manager.set_key(user_id="user123", api_key="sk-xxx", provider="openai")
config_manager.set_key(user_id="user123", api_key="sk-ant-xxx", provider="anthropic")

Note: Keys are stored in memory (not in database). For production multi-server scenarios, consider using Redis.

Environment Variables (Fallback)¶

If no header or user config is provided, CrewAI/LiteLLM will automatically use provider-specific environment variables:

export OPENAI_API_KEY="sk-xxx"
export ANTHROPIC_API_KEY="sk-ant-xxx"
export GOOGLE_API_KEY="xxx"

Examples Auto-initialization¶

When the API server starts, it automatically initializes example tasks if the database is empty. This helps beginners get started quickly.

To manually initialize examples:

# Via CLI
aipartnerupflow examples init

# Or force re-initialization
aipartnerupflow examples init --force

Configuration¶

Environment Variables¶

# Server configuration
export AIPARTNERUPFLOW_HOST=0.0.0.0
export AIPARTNERUPFLOW_PORT=8000

# Database configuration
export AIPARTNERUPFLOW_DATABASE_URL=postgresql://user:pass@localhost/db

# Authentication
export AIPARTNERUPFLOW_JWT_SECRET=your-secret-key

Configuration File¶

Create config.yaml:

server:
  host: 0.0.0.0
  port: 8000
  workers: 4

database:
  url: postgresql://user:pass@localhost/db

auth:
  enabled: true
  jwt_secret: your-secret-key

Production Deployment¶

Using Uvicorn Directly¶

uvicorn aipartnerupflow.api.main:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 4

Using Docker¶

FROM python:3.12
WORKDIR /app
COPY . .
RUN pip install aipartnerupflow[a2a]
CMD ["aipartnerupflow-server", "--host", "0.0.0.0", "--port", "8000"]

Using Systemd¶

Create /etc/systemd/system/aipartnerupflow.service:

[Unit]
Description=aipartnerupflow API Server
After=network.target

[Service]
Type=simple
User=apflow
WorkingDirectory=/opt/aipartnerupflow
ExecStart=/usr/local/bin/aipartnerupflow-server --host 0.0.0.0 --port 8000
Restart=always

[Install]
WantedBy=multi-user.target

Monitoring¶

Health Check¶

curl http://localhost:8000/health

Metrics¶

curl http://localhost:8000/metrics

Troubleshooting¶

Server Won't Start¶

• Check if port is already in use
• Verify database connection
• Check logs for errors

Tasks Not Executing¶

• Verify executor is registered
• Check task name matches executor ID
• Review server logs

Connection Issues¶

• Verify firewall settings
• Check network connectivity
• Ensure server is accessible

MCP Server¶

When started with AIPARTNERUPFLOW_API_PROTOCOL=mcp, the API server exposes task orchestration capabilities as MCP tools and resources.

MCP Tools¶

The MCP server provides 8 tools for task orchestration:

• execute_task - Execute tasks or task trees
• create_task - Create new tasks or task trees
• get_task - Get task details by ID
• update_task - Update existing tasks
• delete_task - Delete tasks (if all pending)
• list_tasks - List tasks with filtering
• get_task_status - Get status of running tasks
• cancel_task - Cancel running tasks

MCP Resources¶

The MCP server provides 2 resource types:

• task://{task_id} - Access individual task data
• tasks:// - Access task list with query parameters (e.g., tasks://?status=running&limit=10)

MCP Endpoints¶

HTTP Mode:

# POST /mcp - JSON-RPC endpoint
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/list",
    "params": {}
  }'

stdio Mode:

# Run as standalone process
python -m aipartnerupflow.api.mcp.server

MCP Usage Example¶

# List available tools
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

# Call a tool
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "execute_task",
    "arguments": {
      "task_id": "task-123"
    }
  }
}

# Read a resource
{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "resources/read",
  "params": {
    "uri": "task://task-123"
  }
}

Next Steps¶

• See HTTP API Reference for complete endpoint documentation
• Check Examples for integration examples
• See Custom Tasks Guide for MCP executor usage