Introducing Opper Agent SDKs - Framework for reliable, headless agents

from opper_agents import Agent, tool

# 1. Define your tools
@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is sunny"

# 2. Create the agent
agent = Agent(
    name="WeatherBot",
    description="Helps with weather queries",
    tools=[get_weather],
    model="cerebras/gpt-oss-120b"
)

# 3. Run it
result = await agent.process("What's the weather in Paris?")

@tool
def analyze_data(data: str) -> dict:
    """Analyze data and return insights."""
    # Your analysis logic here
    return {"insights": ["trend_up", "anomaly_detected"]}

@tool
def send_alert(message: str, priority: str = "medium") -> str:
    """Send an alert notification."""
    # Your alerting logic here
    return f"Alert sent: {message}"

from opper_agents import Agent, mcp, MCPServerConfig

# Configure filesystem access
filesystem_server = MCPServerConfig(
    name="filesystem",
    transport="stdio",
    command="docker",
    args=["run", "-i", "--rm", "-v", f"{os.getcwd()}:/workspace", 
          "node:20", "npx", "-y", "@modelcontextprotocol/server-filesystem", "/workspace"]
)

# Configure database access
sqlite_server = MCPServerConfig(
    name="sqlite", 
    transport="stdio",
    command="uvx",
    args=["mcp-server-sqlite", "--db-path", "./data.db"]
)

# Create agent with MCP tools
agent = Agent(
    name="DataAgent",
    description="Agent with filesystem and database capabilities",
    tools=[mcp(filesystem_server), mcp(sqlite_server)]
)

# For complex reasoning and planning tasks
reasoning_agent = Agent(
    name="ReasoningAgent",
    model="openai/gpt-5",  # Best for complex multi-step reasoning
    tools=[complex_analysis, strategic_planning]
)

# For fast, cost-effective tasks
quick_agent = Agent(
    name="QuickAgent", 
    model="cerebras/gpt-oss-120b",  # Ultra-fast inference
    tools=[simple_classification, data_extraction]
)

# For specialized coding tasks
code_agent = Agent(
    name="CodeAgent",
    model="anthropic/claude-sonnet-4.5",  # Excellent for code generation
    tools=[code_analysis, refactoring]
)

@hook("think_end")
async def on_think_end(
    context: AgentContext, 
    agent: BaseAgent, 
    thought: any
) -> None:
    """Monitor the agent's reasoning process."""
    print(f"Agent thoughts: {thought.reasoning[:100]}...")

# Create agent 
agent = Agent(
    name="MyAgent",
    description="This is my Agent",
    hooks=[on_think_end]
)

@hook("tool_result")
async def clean_tool_output(
    context: AgentContext,
    agent: Agent,
    tool,
    result: ToolResult,
) -> None:
    
    # We use Opper AI to clean the tool output
    from opperai import Opper
    opper = Opper(http_bearer=os.getenv("OPPER_API_KEY"))
    output = opper.call(
        name="clean_tool_output",
        instructions="Clean tool output from irrelevant data and keep only the information that is essential to the task",
        input={
            "goal": context.goal,
            "tool_name": tool.name,
            "tool_result": result.result,
        },
        model="gcp/gemini-flash-latest",
    )
    result.result = output.message

# Create specialized agents
math_agent = Agent(
    name="MathAgent",
    description="Performs mathematical calculations",
    instructions="Always show your work step by step.",
    tools=[calculate, solve_equation]
)

research_agent = Agent(
    name="ResearchAgent", 
    description="Researches and explains concepts",
    instructions="Provide clear, detailed explanations.",
    tools=[search_web, summarize_content]
)

# Use them as tools in a coordinator
coordinator = Agent(
    name="Coordinator",
    description="Delegates tasks to specialized agents",
    tools=[
        math_agent.as_tool(),
        research_agent.as_tool()
    ]
)

# The coordinator can now delegate tasks
result = await coordinator.process(
    "Calculate the compound interest for $1000 at 5% for 10 years, "
    "then research the history of compound interest"
)

# Automatic tracing - no code needed!
agent = Agent(name="MyAgent", tools=[...])
result = await agent.process("My task")

# View traces at https://platform.opper.ai
# See complete execution flow, tool calls, and performance metrics