Python Client

pip install swarms-client

import os
from swarms_client import SwarmsClient
from dotenv import load_dotenv

load_dotenv()

client = SwarmsClient(
    api_key=os.getenv("SWARMS_API_KEY"),
)

# Define your task
task_description = """
Analyze the following patient symptoms and provide ICD code recommendations:
- 45-year-old female with chest pain
- Shortness of breath for 2 days
- Sharp chest pain worsening with deep breathing
- Mild fever (100.2°F) and dry cough
"""

# Create and run a swarm
response = client.swarms.run(
    name="Medical Analysis Swarm",
    description="A swarm that analyzes patient symptoms and provides ICD codes",
    swarm_type="ConcurrentWorkflow",
    task=task_description,
    agents=[
        {
            "agent_name": "Symptom Analyzer",
            "description": "Analyzes patient symptoms and identifies key indicators",
            "system_prompt": "You are a medical expert. Analyze symptoms and identify key medical indicators.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "worker",
            "max_loops": 1,
            "max_tokens": 8192,
            "temperature": 0.3,
        },
        {
            "agent_name": "ICD Code Specialist",
            "description": "Provides appropriate ICD codes based on symptom analysis",
            "system_prompt": "You are an ICD coding specialist. Provide accurate ICD codes with explanations.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "worker",
            "max_loops": 1,
            "max_tokens": 8192,
            "temperature": 0.2,
        }
    ],
)

print(response)

SWARMS_API_KEY=your_api_key_here

from dotenv import load_dotenv
load_dotenv()

client = SwarmsClient()  # Automatically uses SWARMS_API_KEY from environment

client = SwarmsClient(api_key="your_api_key_here")

client = SwarmsClient(
    api_key="your_api_key",
    base_url="https://custom-endpoint.com",  # Optional custom base URL
    timeout=30.0,  # Request timeout in seconds
)

# Create and run a swarm
swarm_response = client.swarms.run(
    name="Data Analysis Swarm",
    description="Analyzes complex datasets",
    swarm_type="SequentialWorkflow",
    task="Analyze the sales data for Q4",
    agents=[...]
)

# Get swarm logs
logs = client.swarms.get_logs()

# Check available swarms
available = client.swarms.check_available()

# List available models
models = client.models.list_available()

# Get model details
for model in models:
    print(f"Model: {model['name']}")
    print(f"Provider: {model['provider']}")
    print(f"Context Length: {model['context_length']}")

# Check API health
health_status = client.health.check()

# Get rate limits
rate_limits = client.client.rate.get_limits()

print(f"API Status: {health_status['status']}")
print(f"Rate Limit: {rate_limits['requests_per_minute']} requests/min")

import asyncio
from swarms_client import AsyncSwarmsClient

async def main():
    client = AsyncSwarmsClient(api_key=os.getenv("SWARMS_API_KEY"))
    
    # Run multiple operations concurrently
    tasks = [
        client.swarms.run(name="Task 1", ...),
        client.swarms.run(name="Task 2", ...),
        client.models.list_available()
    ]
    
    results = await asyncio.gather(*tasks)
    return results

# Run the async function
results = asyncio.run(main())

from swarms_client import SwarmsClient, APIError

try:
    response = client.swarms.run(
        name="Test Swarm",
        description="A test swarm",
        swarm_type="ConcurrentWorkflow",
        task="Simple task",
        agents=[...]
    )
except APIError as e:
    print(f"API Error: {e.message}")
    print(f"Status Code: {e.status_code}")
except Exception as e:
    print(f"Unexpected error: {e}")

# Process multiple tasks
tasks = [
    "Analyze customer feedback for Q1",
    "Generate monthly report summary",
    "Review system performance metrics"
]

responses = []
for task in tasks:
    response = client.swarms.run(
        name=f"Batch Task - {task[:20]}...",
        description="Batch processing task",
        swarm_type="ConcurrentWorkflow",
        task=task,
        agents=[...]
    )
    responses.append(response)

# Quick health check
if client.health.check()['status'] == 'healthy':
    print("API is healthy!")

# Get available models with formatting
models = client.models.list_available()
print(f"Available models: {len(models)}")

# Check rate limits
limits = client.client.rate.get_limits()
print(f"Current usage: {limits['current_usage']}/{limits['requests_per_minute']}")

def safe_api_call(func, *args, **kwargs):
    try:
        return func(*args, **kwargs)
    except APIError as e:
        logger.error(f"API Error: {e.message}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return None

async def main():
    async with AsyncSwarmsClient() as client:
        # Your API calls here
        pass

content_swarm = client.swarms.run(
    name="Content Generation Swarm",
    description="Creates engaging content for social media",
    swarm_type="SequentialWorkflow",
    task="Create a blog post about AI trends in 2024",
    agents=[
        {
            "agent_name": "Research Agent",
            "description": "Researches current AI trends and statistics",
            "system_prompt": "You are a research specialist. Gather current information about AI trends.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "researcher",
            "max_loops": 2,
            "max_tokens": 4096,
            "temperature": 0.3,
        },
        {
            "agent_name": "Writer Agent",
            "description": "Writes engaging blog content based on research",
            "system_prompt": "You are a professional writer. Create engaging blog content.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "writer",
            "max_loops": 1,
            "max_tokens": 8192,
            "temperature": 0.7,
        }
    ],
)

analysis_pipeline = client.swarms.run(
    name="Data Analysis Pipeline",
    description="Comprehensive data analysis workflow",
    swarm_type="SequentialWorkflow",
    task="Analyze quarterly sales data and provide insights",
    agents=[
        {
            "agent_name": "Data Validator",
            "description": "Validates and cleans input data",
            "system_prompt": "You are a data validation expert. Check data quality and format.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "validator",
            "max_loops": 1,
            "max_tokens": 4096,
            "temperature": 0.1,
        },
        {
            "agent_name": "Statistical Analyst",
            "description": "Performs statistical analysis on the data",
            "system_prompt": "You are a statistical analyst. Perform comprehensive data analysis.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "analyst",
            "max_loops": 2,
            "max_tokens": 8192,
            "temperature": 0.2,
        },
        {
            "agent_name": "Insight Generator",
            "description": "Generates actionable insights from analysis",
            "system_prompt": "You are a business analyst. Generate actionable insights.",
            "model_name": "groq/openai/gpt-oss-120b",
            "role": "insights",
            "max_loops": 1,
            "max_tokens": 4096,
            "temperature": 0.5,
        }
    ],
)