Agent

from swarms import Agent

agent = Agent(agent_name="Analyst", model_name="gpt-5.4-mini")

# These are equivalent
agent.get_current_model()
agent.llm_manager.get_current_model()

def run(
    task: Optional[Union[str, Any]] = None,
    img: Optional[str] = None,
    imgs: Optional[List[str]] = None,
    correct_answer: Optional[str] = None,
    streaming_callback: Optional[Callable[[str], None]] = None,
    n: int = 1,
    *args,
    **kwargs
) -> Any

# Basic usage
response = agent.run("Generate a report on financial performance.")

# Single image processing
response = agent.run(
    task="Analyze this image and describe what you see",
    img="path/to/image.jpg"
)

# Multiple image processing
response = agent.run(
    task="Analyze these images and identify common patterns",
    imgs=["image1.jpg", "image2.png", "image3.jpeg"]
)

# Answer validation with retries
response = agent.run(
    task="What is the capital of France?",
    correct_answer="Paris"
)

# Real-time streaming
def streaming_callback(token: str):
    print(token, end="", flush=True)

response = agent.run(
    task="Tell me a long story about space exploration",
    streaming_callback=streaming_callback
)

def __call__(
    task: Optional[str] = None,
    img: Optional[str] = None,
    *args,
    **kwargs
) -> Any

async def arun(
    task: Optional[str] = None,
    img: Optional[str] = None,
    *args,
    **kwargs,
) -> Any

def run_batched(
    tasks: List[str],
    imgs: List[str] = None,
    *args,
    **kwargs,
) -> List[Any]

tasks = [
    "Analyze the financial data for Q1",
    "Generate a summary report for stakeholders",
    "Create recommendations for Q2 planning"
]

batch_results = agent.run_batched(tasks)

for i, (task, result) in enumerate(zip(tasks, batch_results)):
    print(f"Task {i+1}: {task}")
    print(f"Result: {result}\n")

def run_stream(
    task: str,
    img: Optional[str] = None,
    **kwargs,
) -> Iterator[str]

for token in agent.run_stream("Analyse NVDA"):
    print(token, end="", flush=True)

async def arun_stream(
    task: str,
    img: Optional[str] = None,
    **kwargs,
) -> AsyncIterator[str]

import asyncio

async def main():
    async for token in agent.arun_stream("Analyse NVDA"):
        print(token, end="", flush=True)

asyncio.run(main())

def run_concurrent_tasks(
    tasks: List[str],
    *args,
    **kwargs,
) -> List[Any]

def bulk_run(
    inputs: List[Dict[str, Any]],
) -> List[str]

def save(
    file_path: str = None
) -> None

def load(
    file_path: str = None
) -> None

def save_to_yaml(
    file_path: str
) -> None

def to_dict() -> Dict[str, Any]

def to_json(
    indent: int = 4
) -> str

def to_yaml(
    indent: int = 4
) -> str

def to_toml() -> str

def model_dump_json() -> None
def model_dump_yaml() -> None

def add_tool(tool: Callable) -> None
def add_tools(tools: List[Callable]) -> None

def remove_tool(tool: Callable) -> None
def remove_tools(tools: List[Callable]) -> None

def add_memory(message: str) -> None

def talk_to(
    agent: Any,
    task: str,
    img: Optional[str] = None,
    *args,
    **kwargs
) -> Any

def talk_to_multiple_agents(
    agents: List[Union[Any, Callable]],
    task: str,
    *args,
    **kwargs
) -> Any

def receive_message(agent_name: str, task: str, *args, **kwargs) -> Any
def send_agent_message(agent_name: str, message: str, *args, **kwargs) -> str

def reset() -> None

def plan(task: str, *args, **kwargs) -> None

def print_dashboard() -> None

def showcase_config() -> None

def update_system_prompt(system_prompt: str) -> None
def update_max_loops(max_loops: Union[int, str]) -> None
def update_loop_interval(loop_interval: int) -> None

def setup_dynamic_tools(always_loaded: Optional[List[dict]] = None) -> DynamicToolLoader
def defer_tool_schemas(schemas: List[dict]) -> None
def defer_mcp_tools() -> int
def add_mcp_tools_to_memory() -> List[Dict[str, Any]]
def get_all_selected_tools() -> List[str]

def get_llm_parameters() -> str

def get_available_models() -> List[str]
def get_current_model() -> str
def switch_to_next_model() -> bool
def reset_model_index() -> None
def is_fallback_available() -> bool

def handle_skills(task: Optional[str] = None) -> None   # append skills to the system prompt
def load_skills_metadata(skills_dir: str = None) -> List[Dict[str, str]]
def load_full_skill(skill_name: str) -> Optional[str]

def handle_publish_to_marketplace() -> Dict[str, Any]   # requires use_cases

from swarms import Agent
import subprocess

def terminal(code: str):
    """
    Run code in the terminal.

    Args:
        code (str): The code to run in the terminal.

    Returns:
        str: The output of the code.
    """
    out = subprocess.run(code, shell=True, capture_output=True, text=True).stdout
    return str(out)

agent = Agent(
    agent_name="Terminal-Agent",
    model_name="claude-sonnet-4-6",
    tools=[terminal],
    system_prompt="You are an agent that can execute terminal commands.",
)

response = agent.run("List the contents of the current directory")
print(response)

from swarms import Agent
from swarms.prompts.finance_agent_sys_prompt import FINANCIAL_AGENT_SYS_PROMPT
from swarms.utils.str_to_dict import str_to_dict

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_stock_price",
            "description": "Retrieve the current stock price for a specified company.",
            "parameters": {
                "type": "object",
                "properties": {
                    "ticker": {
                        "type": "string",
                        "description": "The stock ticker symbol, e.g. AAPL for Apple Inc.",
                    },
                    "include_history": {
                        "type": "boolean",
                        "description": "Whether to include historical price data.",
                    },
                    "time": {
                        "type": "string",
                        "format": "date-time",
                        "description": "Time for which stock data is requested, in ISO 8601 format.",
                    },
                },
                "required": ["ticker", "include_history", "time"],
            },
        },
    }
]

agent = Agent(
    agent_name="Financial-Analysis-Agent",
    agent_description="Personal finance advisor agent",
    system_prompt=FINANCIAL_AGENT_SYS_PROMPT,
    max_loops=1,
    tools_list_dictionary=tools,
)

out = agent.run("What is the current stock price for Apple Inc. (AAPL)?")
print(out)
print(str_to_dict(out))

from swarms import Agent

# Default: schemas are deferred behind tool_search
agent = Agent(
    agent_name="Toolsmith",
    model_name="gpt-5.4",
    tools=[search_web, read_file, write_file],
)

# Opt out: send every schema on every request
eager_agent = Agent(
    agent_name="Eager-Toolsmith",
    model_name="gpt-5.4",
    tools=[search_web, read_file, write_file],
    dynamic_tools=False,
)

from swarms import Agent


def search_knowledge_base(query: str) -> str:
    """Search the company knowledge base for passages matching a query.

    Args:
        query: What to search for.

    Returns:
        The matching passages, as text.
    """
    # Call whatever store you already run - Chroma, Qdrant, Pinecone, Postgres.
    hits = my_vector_store.query(query, top_k=3)
    return "\n\n".join(h.text for h in hits)


agent = Agent(
    agent_name="Financial-Analysis-Agent",
    model_name="claude-sonnet-4-6",
    system_prompt="You answer using the company knowledge base.",
    tools=[search_knowledge_base],
    max_loops=3,
)

response = agent.run("What are the components of a startup's stock incentive equity plan?")
print(response)

from swarms import Agent

# Opt in: memory survives process restarts
agent = Agent(
    agent_name="Persistent-Agent",
    model_name="gpt-5.4",
    system_prompt="You are a helpful assistant.",
    persistent_memory=True,
)

agent.run("My name is Alice and I work in finance.")
# On the next run — same agent_name, persistent_memory=True again —
# the agent will still know the user's name and role.


# Default: stateless, no cross-session memory
stateless_agent = Agent(
    agent_name="Stateless-Agent",
    model_name="gpt-5.4",
    system_prompt="You are a helpful assistant.",
    # persistent_memory=False is the default
)

stateless_agent.run("Summarise the latest news.")

from swarms import Agent

# Default: automatic summarisation when the window gets full
agent = Agent(
    agent_name="Long-Session-Agent",
    model_name="gpt-5.4",
    system_prompt="You are a research assistant.",
    max_loops=10,
    context_length=8192,
    # context_compression=True is the default
)

agent.run("Deep-dive analysis of transformer architecture papers.")
# If token usage passes ~7,372 tokens (90% of 8192) the compressor
# summarises MEMORY.md in place and continues without interruption.


# Disabled: keep every raw message intact
audit_agent = Agent(
    agent_name="Audit-Agent",
    model_name="gpt-5.4",
    system_prompt="You are a compliance auditor.",
    context_compression=False,
)

from swarms import Agent

# Persistent across runs + automatic context management (production default)
production_agent = Agent(
    agent_name="Production-Agent",
    model_name="gpt-5.4",
    system_prompt="You are a customer support assistant.",
    persistent_memory=True,
    context_compression=True,
    context_length=16000,
)

# Stateless + no compression (CI / unit-test friendly)
test_agent = Agent(
    agent_name="Test-Agent",
    model_name="gpt-5.4",
    system_prompt="You are a helpful assistant.",
    persistent_memory=False,
    context_compression=False,
    print_on=False,
)

from swarms.structs.agent import Agent

research_agent = Agent(
    agent_name="ResearchAgent",
    agent_description="Specializes in researching topics and providing detailed, factual information",
    model_name="gpt-5.4",
    max_loops=1,
    system_prompt="You are a research specialist.",
)

code_agent = Agent(
    agent_name="CodeExpertAgent",
    agent_description="Expert in writing, reviewing, and explaining code",
    model_name="gpt-5.4",
    max_loops=1,
    system_prompt="You are a coding expert.",
)

writing_agent = Agent(
    agent_name="WritingAgent",
    agent_description="Skilled in creative and technical writing",
    model_name="gpt-5.4",
    max_loops=1,
    system_prompt="You are a writing specialist.",
)

coordinator = Agent(
    agent_name="CoordinatorAgent",
    agent_description="Coordinates tasks and delegates to specialized agents",
    model_name="gpt-5.4",
    max_loops=1,
    handoffs=[research_agent, code_agent, writing_agent],
    system_prompt="You are a coordinator agent. Analyze tasks and delegate them to the most appropriate specialized agent.",
    output_type="all",
)

result = coordinator.run(task="Call all the agents and ask them how they are doing")
print(result)

from swarms.structs.agent import Agent

agent = Agent(
    agent_name="Quantitative-Trading-Agent",
    agent_description="Advanced quantitative trading and algorithmic analysis agent",
    model_name="gpt-5.4",
    dynamic_temperature_enabled=True,
    max_loops="auto",
    dynamic_context_window=True,
    output_type="all",
)

out = agent.run(
    "Generate a comprehensive report on the top 5 publicly traded energy stocks. "
    "For each stock include company name, ticker, key financial metrics, and analysis. "
    "Only create 3 subtasks in your plan."
)
print(out)

from swarms.structs.agent import Agent

coordinator = Agent(
    agent_name="Research-Coordinator",
    agent_description="Coordinates complex research by delegating to specialized sub-agents",
    model_name="gpt-5.4",
    max_loops="auto",
    selected_tools="all",
)

task = """
Conduct comprehensive research on three emerging technology trends:
1. Artificial Intelligence in Healthcare
2. Quantum Computing Advances
3. Renewable Energy Innovations

For each topic, create a specialized sub-agent and assign research tasks.
"""

result = coordinator.run(task)
print(result)

tasks = [
    "Analyze the financial data for Q1",
    "Generate a summary report for stakeholders",
    "Create recommendations for Q2 planning"
]

batch_results = agent.run_batched(tasks)

# Batch processing with images
tasks = ["Analyze this chart", "Identify patterns", "Summarize insights"]
images = ["chart1.png", "chart2.png", "chart3.png"]
batch_results = agent.run_batched(tasks, imgs=images)

from swarms import Agent

agent = Agent(
    agent_name="Financial-Analyst",
    model_name="claude-sonnet-4-6",
    max_loops=1,
    system_prompt="You are a financial analyst. Provide detailed, data-driven insights."
)

response = agent.run("Analyze the Q4 2024 revenue trends")
print(response)

from swarms import Agent

agent = Agent(
    model_name="gpt-5.4",
    max_loops=1,
)

response = agent.run("What is the capital of France?")
print(response)

from swarms import Agent

def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

def calculate(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return eval(expression)

agent = Agent(
    agent_name="Research-Agent",
    model_name="claude-sonnet-4-6",
    max_loops=5,
    tools=[search_web, calculate],
    system_prompt="You are a research assistant with web search and calculation abilities."
)

result = agent.run("Search for the population of Tokyo and calculate its growth rate")

agent = Agent(
    agent_name="Vision-Agent",
    model_name="claude-sonnet-4-6",
    multi_modal=True,
    max_loops=1
)

response = agent.run(
    task="Describe what you see in this image and identify any objects",
    img="path/to/image.jpg"
)

image_agent = Agent(
    agent_name="Image-Analysis-Agent",
    system_prompt="You are an expert at analyzing images.",
    multi_modal=True,
)

images = ["product1.jpg", "product2.jpg", "product3.jpg"]
analysis = image_agent.run(
    task="Analyze these product images and identify design patterns",
    imgs=images
)

agent = Agent(
    agent_name="Autonomous-Developer",
    model_name="claude-sonnet-4-6",
    max_loops="auto",
    system_prompt="You are an autonomous software developer."
)

result = agent.run("Build a REST API for a todo application with authentication")
# Agent will:
# 1. Create a plan with subtasks
# 2. Execute each subtask using available tools
# 3. Generate a comprehensive summary

from swarms import Agent

agent = Agent(
    agent_name="Iterative-Reasoning-Agent",
    model_name="gpt-5.4",
    max_loops=3,
    reasoning_prompt_on=True,
    system_prompt="You are an agent that reasons through problems step by step.",
)

response = agent.run("Solve this complex problem step by step: [problem description]")

from swarms import Agent

agent = Agent(
    agent_name="Dynamic-Agent",
    model_name="gpt-5.4",
    dynamic_loops=True,
    system_prompt="You are an adaptive agent that adjusts reasoning depth based on task complexity.",
)

response = agent.run("Analyze this complex scenario and provide insights")

def on_token(token: str):
    print(token, end="", flush=True)

agent = Agent(
    agent_name="Streaming-Agent",
    model_name="claude-sonnet-4-6",
    stream=True,
    streaming_callback=on_token,
    max_loops=1
)

response = agent.run("Write a creative story about space exploration")

from swarms import Agent

agent = Agent(
    model_name="gpt-5.4",
    max_loops=1,
    stream=True,
)

# Each token shows metadata including token count, model info, citations, and usage
agent.run("Tell me a short story about a robot learning to paint.")

agent = Agent(
    agent_name="Reliable-Agent",
    fallback_models=["claude-sonnet-4-6", "gpt-5.4", "gpt-5.4-mini"],
    max_loops=1
)

response = agent.run("Generate a market analysis report")

agent = Agent(
    agent_name="MCP-Agent",
    model_name="claude-sonnet-4-6",
    mcp_url="npx -y @modelcontextprotocol/server-filesystem /path/to/directory",
    max_loops=3
)

result = agent.run("Read the contents of config.json and summarize the settings")

from swarms import Agent

agent = Agent(
    model_name="gpt-5.4",
    mcp_urls=[
        "http://localhost:8000",
        "http://localhost:8001",
    ],
    max_loops=1,
)

response = agent.run("Use tools from both MCP servers")

from swarms import Agent
from swarms.schemas.mcp_schemas import MCPConnection

mcp_config = MCPConnection(
    url="http://localhost:8000",
    name="my_mcp_server",
)

mcp_agent = Agent(
    agent_name="MCP-Enabled-Agent",
    system_prompt="You are an agent with access to external tools via MCP.",
    mcp_config=mcp_config,
    mcp_urls=["http://localhost:8000", "http://localhost:8001"],
    tool_call_summary=True
)

response = mcp_agent.run("Use the available tools to analyze system status")

researcher = Agent(
    agent_name="Researcher",
    model_name="claude-sonnet-4-6",
    system_prompt="You are a research specialist."
)

writer = Agent(
    agent_name="Writer",
    model_name="claude-sonnet-4-6",
    system_prompt="You are a technical writer."
)

coordinator = Agent(
    agent_name="Coordinator",
    model_name="claude-sonnet-4-6",
    max_loops=5,
    handoffs=[researcher, writer],
    system_prompt="You coordinate tasks between research and writing teams."
)

result = coordinator.run("Create a comprehensive report on quantum computing")

from swarms import Agent

agent = Agent(
    agent_name="Interactive-Agent",
    model_name="claude-sonnet-4-6",
    interactive=True,
    system_prompt="You are an interactive agent. Engage in a conversation with the user.",
)

agent.run("Let's start a conversation")

from swarms import Agent

agent = Agent(
    agent_name="Financial-Analysis-Agent",
    system_prompt=None,
    model_name="gpt-5.4",
    max_loops=1,
    auto_generate_prompt=True,
)

agent.run("How can I establish a ROTH IRA to buy stocks and get a tax break?")
print(agent.system_prompt)

from swarms import Agent

agent = Agent(
    model_name="o1-preview",
    reasoning_enabled=True,
    reasoning_effort="high",
    thinking_tokens=10000,
    max_loops=1
)

response = agent.run("Solve this complex mathematical problem step by step")

from swarms import Agent

# Fast mode - optimized for performance
fast_agent = Agent(
    model_name="gpt-5.4",
    mode="fast",
    max_loops=1
)

# Interactive mode - for real-time conversations
interactive_agent = Agent(
    model_name="gpt-5.4",
    mode="interactive",
    max_loops=5
)

# Standard mode - default behavior
standard_agent = Agent(
    model_name="gpt-5.4",
    mode="standard",
    max_loops=1
)

from swarms import Agent

agent = Agent(
    model_name="claude-sonnet-4-6",
    marketplace_prompt_id="550e8400-e29b-41d4-a716-446655440000",
    max_loops=1
)

response = agent.run("Execute the marketplace prompt task")

from swarms import Agent

agent = Agent(
    model_name="gpt-5.4",
    agent_name="Financial-Advisor",
    agent_description="Expert financial advisor agent",
    system_prompt="You are an expert financial advisor...",
    tags=["finance", "advisor"],
    capabilities=["financial_planning", "investment_advice"],
    use_cases=[
        {"title": "Retirement Planning", "description": "Help users plan for retirement"},
        {"title": "Investment Analysis", "description": "Analyze investment opportunities"}
    ],
    publish_to_marketplace=True,
    max_loops=1
)

from swarms import Agent
from swarms.structs.transforms import TransformConfig

transforms = TransformConfig(
    max_tokens=8000,
    strategy="truncate_oldest"
)

agent = Agent(
    model_name="gpt-5.4",
    transforms=transforms,
    context_length=100000,
    max_loops=1
)

response = agent.run("Process this very long conversation history")

from swarms import Agent

agent = Agent(
    model_name="gpt-5.4",
    agent_name="Data-Analysis-Agent",
    capabilities=["data_analysis", "statistics", "visualization"],
    max_loops=1
)

response = agent.run("Analyze this dataset")

# Save the agent state
agent.save('saved_flow.json')

# Load the agent state
agent = Agent(model_name="gpt-5.4", max_loops=5)
agent.load('saved_flow.json')
agent.run("Continue with the task")

from swarms import Agent

agent = Agent(
    model_name="gpt-5.4",
    agent_name="autosave-demo",
    max_loops=5,
    autosave=True,
    verbose=True,
)

response = agent.run("Complete a complex multi-step task")

workspace = agent.workspace.dir
print(f"Files saved to: {workspace}")

# Run several tasks concurrently on a call-scoped thread pool
results = agent.run_concurrent_tasks(["Task A", "Task B"])

# Run multiple tasks concurrently
tasks = [
    {"task": "Task 1"},
    {"task": "Task 2", "img": "path/to/image.jpg"},
    {"task": "Task 3"}
]
responses = agent.bulk_run(tasks)

# Run multiple tasks in batch mode
task_list = ["Analyze data", "Generate report", "Create summary"]
batch_responses = agent.run_batched(task_list)