Python Client

pip install swarms-client

SWARMS_API_KEY=your_api_key_here

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)

import os
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=os.getenv("SWARMS_API_KEY"),
    base_url="https://api.swarms.world",  # 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()

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

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

print(health_status)
print(rate_limits)

import asyncio
import os
from swarms_client import AsyncSwarmsClient

async def main():
    client = AsyncSwarmsClient(api_key=os.getenv("SWARMS_API_KEY"))

    # Run multiple operations concurrently
    results = await asyncio.gather(
        client.swarms.run(
            name="Task 1",
            swarm_type="ConcurrentWorkflow",
            task="Analyze customer feedback for Q1",
            agents=[
                {
                    "agent_name": "Feedback Analyst",
                    "system_prompt": "You analyze customer feedback and summarize key themes.",
                    "model_name": "gpt-4.1",
                }
            ],
        ),
        client.swarms.run(
            name="Task 2",
            swarm_type="ConcurrentWorkflow",
            task="Generate a monthly report summary",
            agents=[
                {
                    "agent_name": "Report Writer",
                    "system_prompt": "You write concise executive report summaries.",
                    "model_name": "gpt-4.1",
                }
            ],
        ),
        client.models.list_available(),
    )
    return results

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

# 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'] == 'ok':
    print("API is healthy!")

# Get available models with formatting
models = client.models.list_available()
print(f"Available models: {models['count']}")

# Check rate limits
limits = client.client.rate.get_limits()
minute = limits['rate_limits']['minute']
print(f"Current usage (last minute): {minute['count']}/{minute['limit']}")
print(f"Tier: {limits['tier']}")

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": 1,
            "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": 1,
            "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,
        }
    ],
)