Swarms API Documentation - Build AI Agents & Multi-Agent Systems

The Swarms API supports real-time streaming responses, allowing you to receive agent outputs as they’re generated. This provides immediate feedback and a better user experience for long-running tasks.

Streaming is enabled by setting "streaming_on": true in your agent configuration.

Quick Start

Enable streaming by adding the streaming_on parameter to your agent configuration:

Python
JavaScript
cURL

import requests
import json
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("SWARMS_API_KEY")
BASE_URL = "https://api.swarms.world"

headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json",
    "Connection": "keep-alive",
    "X-Accel-Buffering": "no"
}

payload = {
    "agent_config": {
        "agent_name": "Research Analyst",
        "model_name": "claude-sonnet-4-20250514",
        "max_tokens": 8192,
        "streaming_on": True
    },
    "task": "What are the key trends in AI development?"
}

response = requests.post(
    f"{BASE_URL}/v1/agent/completions",
    headers=headers,
    json=payload,
    stream=True
)

const fetch = require('node-fetch');
require('dotenv').config();

const API_KEY = process.env.SWARMS_API_KEY;
const BASE_URL = "https://api.swarms.world";

const headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json",
    "Connection": "keep-alive",
    "X-Accel-Buffering": "no"
};

const payload = {
    agent_config: {
        agent_name: "Research Analyst",
        model_name: "claude-sonnet-4-20250514",
        max_tokens: 8192,
        streaming_on: true
    },
    task: "What are the key trends in AI development?"
};

const response = await fetch(`${BASE_URL}/v1/agent/completions`, {
    method: 'POST',
    headers: headers,
    body: JSON.stringify(payload)
});

curl -X POST "https://api.swarms.world/v1/agent/completions" \
  -H "x-api-key: your-api-key" \
  -H "Content-Type: application/json" \
  -H "Connection: keep-alive" \
  -H "X-Accel-Buffering: no" \
  -d '{
    "agent_config": {
      "agent_name": "Research Analyst",
      "model_name": "claude-sonnet-4-20250514",
      "max_tokens": 8192,
      "streaming_on": true
    },
    "task": "What are the key trends in AI development?"
  }' \
  --no-buffer -N

Stream Format

The API uses Server-Sent Events (SSE) format. Each frame is a data: line, optionally preceded by an event: line naming the event type. The very first frame (job metadata) is sent without an event: line — identify it by "type": "metadata" inside the payload:

data: {"job_id": "abc123", "success": true, "name": "Research Analyst", "temperature": 0.7, "stream": true, "type": "metadata"}

event: start
data: {"message": "Starting agent processing..."}

event: chunk
data: {"content": "Based on current research", "timestamp": "2026-07-09T12:00:00Z"}

event: chunk
data: {"content": ", AI development shows", "timestamp": "2026-07-09T12:00:01Z"}

event: usage
data: {"input_tokens": 42, "output_tokens": 108, "total_tokens": 150, "img_cost": 0, "total_cost": 0.00075}

event: end
data: {"job_id": "abc123", "usage": {"...": "..."}, "timestamp": "2026-07-09T12:00:02Z", "complete": true}

event: done
data: {"message": "Agent processing complete"}

On failure, a single error event is sent instead of usage/end/done:

event: error
data: {"error": "AgentCompletionError: ...", "timestamp": "2026-07-09T12:00:02Z"}

Parsing Streams

Here’s how to parse streaming responses in different languages:

Python
JavaScript
Go

def parse_streaming_response(response):
    """Parse streaming response and handle events"""
    full_content = ""
    current_event = None
    
    for line in response.iter_lines():
        if not line:
            continue
            
        line = line.decode("utf-8")
        
        # Parse event type
        if line.startswith("event: "):
            current_event = line[7:].strip()
            continue
        
        # Parse event data
        elif line.startswith("data: "):
            try:
                data = json.loads(line[6:])
                
                # The first frame has no "event:" line — identify it by
                # its "type" field instead.
                if current_event is None and data.get("type") == "metadata":
                    print(f"Job ID: {data.get('job_id')}")
                    print(f"Agent: {data.get('name')}")
                    print("-" * 40)

                elif current_event == "start":
                    print(data.get("message", "Starting..."))

                elif current_event == "chunk":
                    content = data.get("content", "")
                    full_content += content
                    print(content, end="", flush=True)
                
                elif current_event == "usage":
                    print(f"\nTokens used: {data.get('total_tokens')}")
                    print(f"Cost: ${data.get('total_cost', 0):.4f}")

                elif current_event == "end":
                    print(f"\nJob {data.get('job_id')} complete")

                elif current_event == "done":
                    print("\n✅ Complete!")
                
                elif current_event == "error":
                    print(f"\n❌ Error: {data.get('error')}")
                    
            except json.JSONDecodeError:
                continue
    
    return full_content

async function parseStreamingResponse(response) {
    let fullContent = "";
    let currentEvent = null;
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        
        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');
        
        for (const line of lines) {
            if (line.startsWith('event: ')) {
                currentEvent = line.substring(7).trim();
                continue;
            }
            
            if (line.startsWith('data: ')) {
                try {
                    const data = JSON.parse(line.substring(6));
                    
                    // The first frame has no "event: " line — identify it
                    // by its "type" field instead.
                    if (currentEvent === null && data.type === 'metadata') {
                        console.log(`Job ID: ${data.job_id}`);
                        console.log(`Agent: ${data.name}`);
                        console.log("-".repeat(40));
                    } else if (currentEvent === 'start') {
                        console.log(data.message || "Starting...");
                    } else if (currentEvent === 'chunk') {
                        const content = data.content || "";
                        fullContent += content;
                        process.stdout.write(content);
                    } else if (currentEvent === 'usage') {
                        console.log(`\nTokens used: ${data.total_tokens}`);
                        console.log(`Cost: $${data.total_cost?.toFixed(4) || 0}`);
                    } else if (currentEvent === 'end') {
                        console.log(`\nJob ${data.job_id} complete`);
                    } else if (currentEvent === 'done') {
                        console.log("\n✅ Complete!");
                    } else if (currentEvent === 'error') {
                        console.log(`\n❌ Error: ${data.error}`);
                    }
                } catch (e) {
                    // Skip malformed JSON
                }
            }
        }
    }
    
    return fullContent;
}

func parseStreamingResponse(body io.Reader) {
    scanner := bufio.NewScanner(body)
    var currentEvent string
    var fullContent strings.Builder
    
    for scanner.Scan() {
        line := scanner.Text()
        
        if strings.HasPrefix(line, "event: ") {
            currentEvent = strings.TrimSpace(line[7:])
            continue
        }
        
        if strings.HasPrefix(line, "data: ") {
            var data StreamData
            if err := json.Unmarshal([]byte(line[6:]), &data); err != nil {
                continue
            }
            
            switch currentEvent {
            case "":
                // The first frame has no "event: " line — identify it by
                // its "type" field instead.
                if data.Type == "metadata" {
                    fmt.Printf("Job ID: %s\n", data.JobID)
                    fmt.Printf("Agent: %s\n", data.Name)
                    fmt.Println(strings.Repeat("-", 40))
                }
            case "chunk":
                fullContent.WriteString(data.Content)
                fmt.Print(data.Content)
            case "usage":
                fmt.Printf("\nTokens used: %d\n", data.TotalTokens)
                fmt.Printf("Cost: $%.4f\n", data.TotalCost)
            case "done":
                fmt.Println("\n✅ Complete!")
            case "error":
                fmt.Printf("\n❌ Error: %s\n", data.Error)
            }
        }
    }
    
    fmt.Printf("\n📝 Total content: %d characters\n", fullContent.Len())
}

Complete Examples

Python
JavaScript
cURL

import requests
import json
import os
from dotenv import load_dotenv

load_dotenv()

def run_streaming_agent():
    """Complete example of streaming agent request"""
    
    API_KEY = os.getenv("SWARMS_API_KEY")
    BASE_URL = "https://api.swarms.world"
    
    headers = {
        "x-api-key": API_KEY,
        "Content-Type": "application/json",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no"
    }
    
    payload = {
        "agent_config": {
            "agent_name": "Research Analyst",
            "model_name": "claude-sonnet-4-20250514",
            "max_tokens": 8192,
            "streaming_on": True
        },
        "task": "What are the best ways to find samples of diabetes from blood samples?"
    }
    
    print("🚀 Starting streaming request...")
    
    response = requests.post(
        f"{BASE_URL}/v1/agent/completions",
        headers=headers,
        json=payload,
        stream=True,
        timeout=60
    )
    
    if response.status_code != 200:
        print(f"❌ Error: {response.status_code} - {response.text}")
        return
    
    # Parse the streaming response
    full_content = parse_streaming_response(response)
    print(f"\n📝 Total content: {len(full_content)} characters")

# Run the example
if __name__ == "__main__":
    run_streaming_agent()

const fetch = require('node-fetch');
require('dotenv').config();

async function runStreamingAgent() {
    const API_KEY = process.env.SWARMS_API_KEY;
    const BASE_URL = "https://api.swarms.world";
    
    const headers = {
        "x-api-key": API_KEY,
        "Content-Type": "application/json",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no"
    };
    
    const payload = {
        agent_config: {
            agent_name: "Research Analyst",
            model_name: "claude-sonnet-4-20250514",
            max_tokens: 8192,
            streaming_on: true
        },
        task: "What are the best ways to find samples of diabetes from blood samples?"
    };
    
    console.log("🚀 Starting streaming request...");
    
    try {
        const response = await fetch(`${BASE_URL}/v1/agent/completions`, {
            method: 'POST',
            headers: headers,
            body: JSON.stringify(payload)
        });
        
        if (!response.ok) {
            console.error(`❌ Error: ${response.status} - ${await response.text()}`);
            return;
        }
        
        // Parse streaming response
        const fullContent = await parseStreamingResponse(response);
        console.log(`\n📝 Total content: ${fullContent.length} characters`);
        
    } catch (error) {
        console.error("Request failed:", error);
    }
}

// Run the example
runStreamingAgent();

#!/bin/bash

API_KEY="your-api-key-here"
BASE_URL="https://api.swarms.world"

echo "🚀 Starting streaming request..."

curl -X POST "${BASE_URL}/v1/agent/completions" \
  -H "x-api-key: ${API_KEY}" \
  -H "Content-Type: application/json" \
  -H "Connection: keep-alive" \
  -H "X-Accel-Buffering: no" \
  -d '{
    "agent_config": {
      "agent_name": "Research Analyst",
      "model_name": "claude-sonnet-4-20250514",
      "max_tokens": 8192,
      "streaming_on": true
    },
    "task": "What are the best ways to find samples of diabetes from blood samples?"
  }' \
  --no-buffer \
  -N

Event Types

Event	Description	Data Fields
(none — first frame)	Job metadata, sent before any `event:` line	`job_id`, `success`, `name`, `description`, `temperature`, `timestamp`, `stream`, `type` (`"metadata"`)
`start`	Agent processing has begun	`message`
`chunk`	Content piece	`content`, `timestamp`
`usage`	Token usage and cost, sent once near the end	`input_tokens`, `output_tokens`, `total_tokens`, `img_cost`, `total_cost`
`end`	Final job metadata	`job_id`, `usage`, `timestamp`, `complete`
`done`	Stream finished	`message`
`error`	Error info	`error`, `timestamp`

Best Practices

Error Handling

Always handle potential errors in your stream processing:

try:
    response = requests.post(url, json=payload, stream=True, timeout=60)
    if response.status_code != 200:
        print(f"Error: {response.status_code} - {response.text}")
        return
    
    full_content = parse_streaming_response(response)
    
except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")
except json.JSONDecodeError as e:
    print(f"JSON decode error: {e}")

Timeout Management

Set appropriate timeouts for your use case:

# For quick responses
response = requests.post(url, json=payload, stream=True, timeout=30)

# For long-running tasks  
response = requests.post(url, json=payload, stream=True, timeout=300)

Benefits

Real-time Feedback: See results as they’re generated
Better UX: Reduced perceived latency
Progress Tracking: Monitor long-running operations
Error Handling: Immediate error feedback

Troubleshooting

Connection Timeouts

Increase timeout values for long-running tasks. Set appropriate timeouts based on your expected response time.

JSON Decode Errors

Handle malformed data gracefully by wrapping JSON parsing in try-catch blocks.

Incomplete Streams

Always check for done or error events to ensure the stream completed successfully.

Memory Usage

Process chunks incrementally for large responses to avoid memory issues.

Debug Mode

Enable debug logging to troubleshoot stream issues:

import logging
logging.basicConfig(level=logging.DEBUG)

Getting Started

Clients

Agent Completions

Multi-Agent

Batch

Capabilities

Resources

Swarms Cloud

Streaming

Quick Start

Stream Format

Parsing Streams

Complete Examples

Event Types

Best Practices

Error Handling

Timeout Management

Benefits

Troubleshooting

Debug Mode

​Quick Start

​Stream Format

​Parsing Streams

​Complete Examples

​Event Types

​Best Practices

​Error Handling

​Timeout Management

​Benefits

​Troubleshooting

​Debug Mode

Quick Start

Stream Format

Parsing Streams

Complete Examples

Event Types

Best Practices

Error Handling

Timeout Management

Benefits

Troubleshooting

Debug Mode