swarms/docs/swarms_cloud/python_client.md

Swarms Cloud API Client Documentation

Introduction

The Swarms Cloud API client is a production-grade Python package for interacting with the Swarms API. It provides both synchronous and asynchronous interfaces, making it suitable for a wide range of applications from simple scripts to high-performance, scalable services.

Key features include:

• Connection pooling and efficient session management
• Automatic retries with exponential backoff
• Circuit breaker pattern for improved reliability
• In-memory caching for frequently accessed resources
• Comprehensive error handling with detailed exceptions
• Full support for asynchronous operations
• Type checking with Pydantic

This documentation covers all available client methods with detailed descriptions, parameter references, and usage examples.

Installation

pip install swarms-client

Authentication

To use the Swarms API, you need an API key. You can obtain your API key from the Swarms Platform API Keys page.

Client Initialization

The SwarmsClient is the main entry point for interacting with the Swarms API. It can be initialized with various configuration options to customize its behavior.

from swarms_client import SwarmsClient

# Initialize with default settings
client = SwarmsClient(api_key="your-api-key")

# Or with custom settings
client = SwarmsClient(
    api_key="your-api-key",
    base_url="https://swarms-api-285321057562.us-east1.run.app",
    timeout=60,
    max_retries=3,
    retry_delay=1,
    log_level="INFO",
    pool_connections=100,
    pool_maxsize=100,
    keep_alive_timeout=5,
    max_concurrent_requests=100,
    circuit_breaker_threshold=5,
    circuit_breaker_timeout=60,
    enable_cache=True
)

Parameters

Parameter	Type	Default	Description
api_key	str	Environment variable SWARMS_API_KEY	API key for authentication
base_url	str	"https://swarms-api-285321057562.us-east1.run.app"	Base URL for the API
timeout	int	60	Timeout for API requests in seconds
max_retries	int	3	Maximum number of retry attempts for failed requests
retry_delay	int	1	Initial delay between retries in seconds (uses exponential backoff)
log_level	str	"INFO"	Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
pool_connections	int	100	Number of connection pools to cache
pool_maxsize	int	100	Maximum number of connections to save in the pool
keep_alive_timeout	int	5	Keep-alive timeout for connections in seconds
max_concurrent_requests	int	100	Maximum number of concurrent requests
circuit_breaker_threshold	int	5	Failure threshold for the circuit breaker
circuit_breaker_timeout	int	60	Reset timeout for the circuit breaker in seconds
enable_cache	bool	True	Whether to enable in-memory caching

Client Methods

clear_cache

Clears the in-memory cache used for caching API responses.

client.clear_cache()

Agent Resource

The Agent resource provides methods for creating and managing agent completions.

create

Creates an agent completion.

response = client.agent.create(
    agent_config={
        "agent_name": "Researcher",
        "description": "Conducts in-depth research on topics",
        "model_name": "gpt-4o",
        "temperature": 0.7
    },
    task="Research the latest advancements in quantum computing and summarize the key findings"
)

print(f"Agent ID: {response.id}")
print(f"Output: {response.outputs}")

Parameters

Parameter	Type	Required	Description
agent_config	dict or AgentSpec	Yes	Configuration for the agent
task	str	Yes	The task for the agent to complete
history	dict	No	Optional conversation history

The agent_config parameter can include the following fields:

Field	Type	Default	Description
agent_name	str	Required	Name of the agent
description	str	None	Description of the agent's purpose
system_prompt	str	None	System prompt to guide the agent's behavior
model_name	str	"gpt-4o-mini"	Name of the model to use
auto_generate_prompt	bool	False	Whether to automatically generate a prompt
max_tokens	int	8192	Maximum tokens in the response
temperature	float	0.5	Temperature for sampling (0-1)
role	str	None	Role of the agent
max_loops	int	1	Maximum number of reasoning loops
tools_dictionary	List[Dict]	None	Tools available to the agent

Returns

AgentCompletionResponse object with the following properties:

• id: Unique identifier for the completion
• success: Whether the completion was successful
• name: Name of the agent
• description: Description of the agent
• temperature: Temperature used for the completion
• outputs: Output from the agent
• usage: Token usage information
• timestamp: Timestamp of the completion

create_batch

Creates multiple agent completions in batch.

responses = client.agent.create_batch([
    {
        "agent_config": {
            "agent_name": "Researcher",
            "model_name": "gpt-4o-mini",
            "temperature": 0.5
        },
        "task": "Summarize the latest quantum computing research"
    },
    {
        "agent_config": {
            "agent_name": "Writer",
            "model_name": "gpt-4o",
            "temperature": 0.7
        },
        "task": "Write a blog post about AI safety"
    }
])

for i, response in enumerate(responses):
    print(f"Agent {i+1} ID: {response.id}")
    print(f"Output: {response.outputs}")
    print("---")

Parameters

Parameter	Type	Required	Description
completions	List[Dict or AgentCompletion]	Yes	List of agent completion requests

Each item in the completions list should have the same structure as the parameters for the create method.

Returns

List of AgentCompletionResponse objects with the same properties as the return value of the create method.

acreate

Creates an agent completion asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient(api_key="your-api-key") as client:
        response = await client.agent.acreate(
            agent_config={
                "agent_name": "Researcher",
                "description": "Conducts in-depth research",
                "model_name": "gpt-4o"
            },
            task="Research the impact of quantum computing on cryptography"
        )
        
        print(f"Agent ID: {response.id}")
        print(f"Output: {response.outputs}")

asyncio.run(main())

Parameters

Same as the create method.

Returns

Same as the create method.

acreate_batch

Creates multiple agent completions in batch asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient(api_key="your-api-key") as client:
        responses = await client.agent.acreate_batch([
            {
                "agent_config": {
                    "agent_name": "Researcher",
                    "model_name": "gpt-4o-mini"
                },
                "task": "Summarize the latest quantum computing research"
            },
            {
                "agent_config": {
                    "agent_name": "Writer",
                    "model_name": "gpt-4o"
                },
                "task": "Write a blog post about AI safety"
            }
        ])
        
        for i, response in enumerate(responses):
            print(f"Agent {i+1} ID: {response.id}")
            print(f"Output: {response.outputs}")
            print("---")

asyncio.run(main())

Parameters

Same as the create_batch method.

Returns

Same as the create_batch method.

Swarm Resource

The Swarm resource provides methods for creating and managing swarm completions.

create

Creates a swarm completion.

response = client.swarm.create(
    name="Research Swarm",
    description="A swarm for research tasks",
    swarm_type="SequentialWorkflow",
    task="Research quantum computing advances in 2024 and summarize the key findings",
    agents=[
        {
            "agent_name": "Researcher",
            "description": "Conducts in-depth research",
            "model_name": "gpt-4o",
            "temperature": 0.5
        },
        {
            "agent_name": "Critic",
            "description": "Evaluates arguments for flaws",
            "model_name": "gpt-4o-mini",
            "temperature": 0.3
        }
    ],
    max_loops=3,
    return_history=True
)

print(f"Job ID: {response.job_id}")
print(f"Status: {response.status}")
print(f"Output: {response.output}")

Parameters

Parameter	Type	Required	Description
name	str	No	Name of the swarm
description	str	No	Description of the swarm
agents	List[Dict or AgentSpec]	No	List of agent specifications
max_loops	int	No	Maximum number of loops (default: 1)
swarm_type	str	No	Type of swarm (see available types)
task	str	Conditional	The task to complete (required if tasks and messages are not provided)
tasks	List[str]	Conditional	List of tasks for batch processing (required if task and messages are not provided)
messages	List[Dict]	Conditional	List of messages to process (required if task and tasks are not provided)
return_history	bool	No	Whether to return the execution history (default: True)
rules	str	No	Rules for the swarm
schedule	Dict	No	Schedule specification for delayed execution
stream	bool	No	Whether to stream the response (default: False)
service_tier	str	No	Service tier ('standard' or 'flex', default: 'standard')

Returns

SwarmCompletionResponse object with the following properties:

• job_id: Unique identifier for the job
• status: Status of the job
• swarm_name: Name of the swarm
• description: Description of the swarm
• swarm_type: Type of swarm used
• output: Output from the swarm
• number_of_agents: Number of agents in the swarm
• service_tier: Service tier used
• tasks: List of tasks processed (if applicable)
• messages: List of messages processed (if applicable)

create_batch

Creates multiple swarm completions in batch.

responses = client.swarm.create_batch([
    {
        "name": "Research Swarm",
        "swarm_type": "auto",
        "task": "Research quantum computing advances",
        "agents": [
            {"agent_name": "Researcher", "model_name": "gpt-4o"}
        ]
    },
    {
        "name": "Writing Swarm",
        "swarm_type": "SequentialWorkflow",
        "task": "Write a blog post about AI safety",
        "agents": [
            {"agent_name": "Writer", "model_name": "gpt-4o"},
            {"agent_name": "Editor", "model_name": "gpt-4o-mini"}
        ]
    }
])

for i, response in enumerate(responses):
    print(f"Swarm {i+1} Job ID: {response.job_id}")
    print(f"Status: {response.status}")
    print(f"Output: {response.output}")
    print("---")

Parameters

Parameter	Type	Required	Description
swarms	List[Dict or SwarmSpec]	Yes	List of swarm specifications

Each item in the swarms list should have the same structure as the parameters for the create method.

Returns

List of SwarmCompletionResponse objects with the same properties as the return value of the create method.

list_types

Lists available swarm types.

response = client.swarm.list_types()

print(f"Available swarm types:")
for swarm_type in response.swarm_types:
    print(f"- {swarm_type}")

Returns

SwarmTypesResponse object with the following properties:

• success: Whether the request was successful
• swarm_types: List of available swarm types

alist_types

Lists available swarm types asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient(api_key="your-api-key") as client:
        response = await client.swarm.alist_types()
        
        print(f"Available swarm types:")
        for swarm_type in response.swarm_types:
            print(f"- {swarm_type}")

asyncio.run(main())

Returns

Same as the list_types method.

acreate

Creates a swarm completion asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient(api_key="your-api-key") as client:
        response = await client.swarm.acreate(
            name="Research Swarm",
            swarm_type="SequentialWorkflow",
            task="Research quantum computing advances in 2024",
            agents=[
                {
                    "agent_name": "Researcher",
                    "description": "Conducts in-depth research",
                    "model_name": "gpt-4o"
                },
                {
                    "agent_name": "Critic",
                    "description": "Evaluates arguments for flaws",
                    "model_name": "gpt-4o-mini"
                }
            ]
        )
        
        print(f"Job ID: {response.job_id}")
        print(f"Status: {response.status}")
        print(f"Output: {response.output}")

asyncio.run(main())

Parameters

Same as the create method.

Returns

Same as the create method.

acreate_batch

Creates multiple swarm completions in batch asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient(api_key="your-api-key") as client:
        responses = await client.swarm.acreate_batch([
            {
                "name": "Research Swarm",
                "swarm_type": "auto",
                "task": "Research quantum computing",
                "agents": [
                    {"agent_name": "Researcher", "model_name": "gpt-4o"}
                ]
            },
            {
                "name": "Writing Swarm",
                "swarm_type": "SequentialWorkflow",
                "task": "Write a blog post about AI safety",
                "agents": [
                    {"agent_name": "Writer", "model_name": "gpt-4o"}
                ]
            }
        ])
        
        for i, response in enumerate(responses):
            print(f"Swarm {i+1} Job ID: {response.job_id}")
            print(f"Status: {response.status}")
            print(f"Output: {response.output}")
            print("---")

asyncio.run(main())

Parameters

Same as the create_batch method.

Returns

Same as the create_batch method.

Models Resource

The Models resource provides methods for retrieving information about available models.

list

Lists available models.

response = client.models.list()

print(f"Available models:")
for model in response.models:
    print(f"- {model}")

Returns

ModelsResponse object with the following properties:

• success: Whether the request was successful
• models: List of available model names

alist

Lists available models asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient(api_key="your-api-key") as client:
        response = await client.models.alist()
        
        print(f"Available models:")
        for model in response.models:
            print(f"- {model}")

asyncio.run(main())

Returns

Same as the list method.

Logs Resource

The Logs resource provides methods for retrieving API request logs.

list

Lists API request logs.

response = client.logs.list()

print(f"Found {response.count} logs:")
for log in response.logs:
    print(f"- ID: {log.id}, Created at: {log.created_at}")
    print(f"  Data: {log.data}")

Returns

LogsResponse object with the following properties:

• status: Status of the request
• count: Number of logs
• logs: List of log entries
• timestamp: Timestamp of the request

Each log entry is a LogEntry object with the following properties:

• id: Unique identifier for the log entry
• api_key: API key used for the request
• data: Request data
• created_at: Timestamp when the log entry was created

alist

Lists API request logs asynchronously.

import asyncio
from swarms_client import SwarmsClient

async def main():
    async with SwarmsClient() as client:
        response = await client.logs.alist()
        
        print(f"Found {response.count} logs:")
        for log in response.logs:
            print(f"- ID: {log.id}, Created at: {log.created_at}")
            print(f"  Data: {log.data}")

asyncio.run(main())

Returns

Same as the list method.

Error Handling

The Swarms API client provides detailed error handling with specific exception types for different error scenarios. All exceptions inherit from the base SwarmsError class.

from swarms_client import SwarmsClient, SwarmsError, AuthenticationError, RateLimitError, APIError

try:
    client = SwarmsClient(api_key="invalid-api-key")
    response = client.agent.create(
        agent_config={"agent_name": "Researcher", "model_name": "gpt-4o"},
        task="Research quantum computing"
    )
except AuthenticationError as e:
    print(f"Authentication error: {e}")
except RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
except APIError as e:
    print(f"API error: {e}")
except SwarmsError as e:
    print(f"Swarms error: {e}")

Exception Types

Exception	Description
SwarmsError	Base exception for all Swarms API errors
AuthenticationError	Raised when there's an issue with authentication
RateLimitError	Raised when the rate limit is exceeded
APIError	Raised when the API returns an error
InvalidRequestError	Raised when the request is invalid
InsufficientCreditsError	Raised when the user doesn't have enough credits
TimeoutError	Raised when a request times out
NetworkError	Raised when there's a network issue

Advanced Features

Connection Pooling

The Swarms API client uses connection pooling to efficiently manage HTTP connections, which can significantly improve performance when making multiple requests.

client = SwarmsClient(
    api_key="your-api-key",
    pool_connections=100,  # Number of connection pools to cache
    pool_maxsize=100,      # Maximum number of connections to save in the pool
    keep_alive_timeout=5   # Keep-alive timeout for connections in seconds
)

Circuit Breaker Pattern

The client implements the circuit breaker pattern to prevent cascading failures when the API is experiencing issues.

client = SwarmsClient(
    api_key="your-api-key",
    circuit_breaker_threshold=5,  # Number of failures before the circuit opens
    circuit_breaker_timeout=60    # Time in seconds before attempting to close the circuit
)

Caching

The client includes in-memory caching for frequently accessed resources to reduce API calls and improve performance.

client = SwarmsClient(
    api_key="your-api-key",
    enable_cache=True  # Enable in-memory caching
)

# Clear the cache manually if needed
client.clear_cache()

Complete Example

Here's a complete example that demonstrates how to use the Swarms API client to create a research swarm and process its output:

import os
from swarms_client import SwarmsClient
from dotenv import load_dotenv

# Load API key from environment
load_dotenv()
api_key = os.getenv("SWARMS_API_KEY")

# Initialize client
client = SwarmsClient(api_key=api_key)

# Create a research swarm
try:
    # Define the agents
    researcher = {
        "agent_name": "Researcher",
        "description": "Conducts thorough research on specified topics",
        "model_name": "gpt-4o",
        "temperature": 0.5,
        "system_prompt": "You are a diligent researcher focused on finding accurate and comprehensive information."
    }
    
    analyst = {
        "agent_name": "Analyst",
        "description": "Analyzes research findings and identifies key insights",
        "model_name": "gpt-4o",
        "temperature": 0.3,
        "system_prompt": "You are an insightful analyst who can identify patterns and extract meaningful insights from research data."
    }
    
    summarizer = {
        "agent_name": "Summarizer",
        "description": "Creates concise summaries of complex information",
        "model_name": "gpt-4o-mini",
        "temperature": 0.4,
        "system_prompt": "You specialize in distilling complex information into clear, concise summaries."
    }
    
    # Create the swarm
    response = client.swarm.create(
        name="Quantum Computing Research Swarm",
        description="A swarm for researching and analyzing quantum computing advancements",
        swarm_type="SequentialWorkflow",
        task="Research the latest advancements in quantum computing in 2024, analyze their potential impact on cryptography and data security, and provide a concise summary of the findings.",
        agents=[researcher, analyst, summarizer],
        max_loops=2,
        return_history=True
    )
    
    # Process the response
    print(f"Job ID: {response.job_id}")
    print(f"Status: {response.status}")
    print(f"Number of agents: {response.number_of_agents}")
    print(f"Swarm type: {response.swarm_type}")
    
    # Print the output
    if "final_output" in response.output:
        print("\nFinal Output:")
        print(response.output["final_output"])
    else:
        print("\nOutput:")
        print(response.output)
    
    # Access agent-specific outputs if available
    if "agent_outputs" in response.output:
        print("\nAgent Outputs:")
        for agent, output in response.output["agent_outputs"].items():
            print(f"\n{agent}:")
            print(output)

except Exception as e:
    print(f"Error: {e}")

This example creates a sequential workflow swarm with three agents to research quantum computing, analyze the findings, and create a summary of the results.