Openai Agents Sdk Skill

OpenAI Agents SDK Skill

Use this skill when implementing AI assistants using the OpenAI Agents SDK with function tools, MCP server integration, and context management.

When to Use

• Creating AI agents with function tools and structured outputs
• Integrating Model Context Protocol (MCP) servers
• Managing conversational history and context
• Building tool-enabled AI assistants
• Implementing dynamic tool filtering and security

Core Components

1. Agent Creation with OpenRouter

# app/agents/todo_agent.py
import os
from agents import Agent, Runner
from agents import AsyncOpenAI, OpenAIChatCompletionsModel
from agents.run import RunConfig
from pydantic import BaseModel, Field
from typing import Annotated
import asyncio

class TodoAgent:
    """AI agent for managing todo tasks with OpenAI Agents SDK using OpenRouter."""

    def __init__(self, user_id: str):
        self.user_id = user_id
        
        # Setup OpenRouter client using specific configuration
        openrouter_api_key = os.getenv("OPENROUTER_API_KEY")
        openrouter_base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
        openrouter_model = os.getenv("OPENROUTER_MODEL", "mistralai/devstral-2512:free")
        
        if not openrouter_api_key:
            raise ValueError("OPENROUTER_API_KEY is not set in environment variables")
        
        external_client = AsyncOpenAI(
            api_key=openrouter_api_key,
            base_url=openrouter_base_url,
        )

        # Define the model to use (using available OpenRouter models)
        model = OpenAIChatCompletionsModel(
            model=openrouter_model,  # Using mistralai/devstral-2512:free as specified
            openai_client=external_client
        )

        # Create run configuration with OpenRouter
        self.config = RunConfig(
            model=model,
            model_provider=external_client,
            # Add other configuration as needed
        )

        # Create the agent with tools
        self.agent = Agent(
            name="Todo Assistant",
            instructions="You are a helpful assistant for managing todo tasks. Use available tools to add, list, complete, update, and delete tasks.",
            tools=self.get_tools()
        )

    def get_tools(self):
        """Define and return agent tools."""
        return [
            self.add_task_tool,
            self.list_tasks_tool,
            self.complete_task_tool,
            self.update_task_tool,
            self.delete_task_tool
        ]

    async def run(self, message: str):
        """Run the agent with user message using OpenRouter configuration."""
        return await Runner.run(
            self.agent,
            message,
            config=self.config
        )

2. Function Tools with Pydantic Models

# app/agents/tools.py
from agents import function_tool
from pydantic import BaseModel, Field
from typing import Annotated, Optional
import asyncio

# Define structured output models
class Task(BaseModel):
    id: int = Field(description="Unique task identifier")
    title: str = Field(description="Task title")
    description: Optional[str] = Field(description="Task description")
    completed: bool = Field(description="Whether task is completed")
    priority: str = Field(description="Task priority: low, medium, high")

class AddTaskParams(BaseModel):
    title: str = Field(description="Task title")
    description: Optional[str] = Field(description="Task description")
    priority: str = Field(description="Task priority: low, medium, high")

# Create function tools with decorators
@function_tool
def add_task(params: AddTaskParams) -> Task:
    """Add a new task to the user's list."""
    # In production, this would call your task service
    return Task(
        id=123,
        title=params.title,
        description=params.description,
        completed=False,
        priority=params.priority or "medium"
    )

@function_tool
def list_tasks(
    status: Optional[str] = "all",
    priority: Optional[str] = None
) -> list[Task]:
    """List user's tasks with optional filtering."""
    # In production, this would fetch from your database
    return [
        Task(id=1, title="Buy groceries", completed=False, priority="medium"),
        Task(id=2, title="Call doctor", completed=True, priority="high")
    ]

@function_tool
def complete_task(task_id: int) -> Task:
    """Mark a task as completed."""
    # In production, this would update your database
    return Task(
        id=task_id,
        title="Sample task",
        completed=True,
        priority="medium"
    )

@function_tool
def update_task(
    task_id: int,
    title: Optional[str] = None,
    description: Optional[str] = None
) -> Task:
    """Update an existing task."""
    # In production, this would update your database
    return Task(
        id=task_id,
        title=title or f"Updated task {task_id}",
        description=description,
        completed=False,
        priority="medium"
    )

@function_tool
def delete_task(task_id: int) -> dict:
    """Delete a task by ID."""
    # In production, this would delete from your database
    return {"success": True, "deleted_task_id": task_id}

3. MCP Server Integration

# app/agents/mcp_integration.py
from agents.mcp import MCPServerStdio
from agents import Agent
import asyncio
import os
from pathlib import Path

class MCPIntegration:
    """Handle MCP server integration for the Todo agent."""

    def __init__(self, user_workspace: str):
        self.user_workspace = user_workspace
        self.server = None

    async def setup_server(self):
        """Setup MCP server connection."""
        # Create MCP server for file operations
        self.server = MCPServerStdio(
            name="Todo Workspace Server",
            params={
                "command": "python",
                "args": ["-m", "app.mcp.todo_server", self.user_workspace]
            }
        )

        # Connect to the server
        await self.server.connect()

        return self.server

    async def get_agent_with_mcp(self):
        """Create agent with MCP server integration."""
        # Setup MCP server
        mcp_server = await self.setup_server()

        # Create agent with MCP server
        agent = Agent(
            name="Todo Assistant with MCP",
            instructions="Use MCP tools to access and manage user's todo workspace.",
            mcp_servers=[mcp_server]
        )

        return agent

    async def cleanup(self):
        """Clean up MCP server connection."""
        if self.server:
            await self.server.cleanup()

4. Context Management and History

# app/agents/context_manager.py
from agents import user, assistant, system
from typing import List, Dict, Any
from datetime import datetime

class ContextManager:
    """Manage conversational history and context for the agent."""

    def __init__(self, user_id: str):
        self.user_id = user_id
        self.history = self._initialize_history()

    def _initialize_history(self) -> List[Dict[str, Any]]:
        """Initialize conversation history with system message."""
        return [
            system(f"You are a helpful todo assistant for user {self.user_id}. Today is {datetime.now().strftime('%Y-%m-%d')}."),
        ]

    def add_user_message(self, message: str):
        """Add user message to history."""
        self.history.append(user(message))

    def add_assistant_response(self, response: str):
        """Add assistant response to history."""
        self.history.append(assistant(response))

    def get_context(self) -> List[Dict[str, Any]]:
        """Get current conversation context."""
        return self.history

    def clear_history(self):
        """Clear conversation history."""
        self.history = self._initialize_history()

    def add_memory(self, key: str, value: str):
        """Add contextual memory to the conversation."""
        self.history.append(system(f"[Memory: {key}={value}]"))

5. Complete Agent Implementation

# app/agents/todo_assistant.py
from agents import Agent, Runner, function_tool
from agents.mcp import MCPServerStdio
from agents import AsyncOpenAI, OpenAIChatCompletionsModel
from agents.run import RunConfig
from pydantic import BaseModel, Field
from typing import Optional, List
import os
import asyncio

class TodoAssistant:
    """Complete implementation of a Todo assistant using OpenAI Agents SDK with OpenRouter."""

    def __init__(self, user_id: str):
        self.user_id = user_id
        self.context_manager = ContextManager(user_id)
        self.mcp_server = None
        
        # Setup OpenRouter client using specific configuration
        openrouter_api_key = os.getenv("OPENROUTER_API_KEY")
        openrouter_base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
        openrouter_model = os.getenv("OPENROUTER_MODEL", "mistralai/devstral-2512:free")
        
        if not openrouter_api_key:
            raise ValueError("OPENROUTER_API_KEY is not set in environment variables")
        
        external_client = AsyncOpenAI(
            api_key=openrouter_api_key,
            base_url=openrouter_base_url,
        )

        # Define the model to use (using available OpenRouter models)
        model = OpenAIChatCompletionsModel(
            model=openrouter_model,  # Using mistralai/devstral-2512:free as specified
            openai_client=external_client
        )

        # Create run configuration with OpenRouter
        self.config = RunConfig(
            model=model,
            model_provider=external_client,
            # Add other configuration as needed
        )

    async def setup_mcp_server(self):
        """Setup MCP server for extended functionality."""
        # Create MCP server for advanced operations
        self.mcp_server = MCPServerStdio(
            name="Todo MCP Server",
            params={
                "command": "python",
                "args": ["-m", "app.mcp.todo_operations"]
            }
        )
        await self.mcp_server.connect()

    def create_agent(self) -> Agent:
        """Create the Todo assistant agent."""
        # Prepare tools
        tools = [
            self.add_task_tool,
            self.list_tasks_tool,
            self.complete_task_tool,
            self.update_task_tool,
            self.delete_task_tool,
        ]

        # Create agent with MCP server if available
        agent_kwargs = {
            "name": "Todo Assistant",
            "instructions": """
            You are a helpful assistant for managing todo tasks.
            - Always use the appropriate tools for task operations
            - Confirm actions before executing destructive operations (delete, complete)
            - Provide clear, friendly responses
            - Maintain context throughout the conversation
            """,
            "tools": tools
        }

        if self.mcp_server:
            agent_kwargs["mcp_servers"] = [self.mcp_server]

        return Agent(**agent_kwargs)

    async def process_message(self, message: str) -> Dict[str, Any]:
        """Process user message through the agent."""
        # Add user message to context
        self.context_manager.add_user_message(message)

        # Create agent
        agent = self.create_agent()

        # Run agent with current context
        result = await Runner.run(
            agent,
            self.context_manager.get_context(),
            config=self.config  # Use OpenRouter configuration
        )

        # Add assistant response to context
        self.context_manager.add_assistant_response(result.final_output)

        return {
            "response": result.final_output,
            "tool_calls": result.tool_calls,
            "context": self.context_manager.get_context()
        }

    # Define function tools
    @function_tool
    def add_task_tool(
        self,
        title: str,
        description: Optional[str] = None,
        priority: str = "medium"
    ) -> dict:
        """Add a new task to the user's list."""
        # In production, this would call your task service
        return {
            "success": True,
            "task_id": 123,
            "title": title,
            "description": description,
            "priority": priority,
            "message": f"Added task: {title}"
        }

    @function_tool
    def list_tasks_tool(
        self,
        status: Optional[str] = "all",
        priority: Optional[str] = None,
        limit: Optional[int] = 10
    ) -> dict:
        """List user's tasks with optional filtering."""
        # In production, this would fetch from your database
        tasks = [
            {"id": 1, "title": "Buy groceries", "completed": False, "priority": "medium"},
            {"id": 2, "title": "Call doctor", "completed": True, "priority": "high"}
        ]

        return {
            "tasks": tasks,
            "count": len(tasks),
            "filters_applied": {"status": status, "priority": priority}
        }

    @function_tool
    def complete_task_tool(self, task_id: int) -> dict:
        """Mark a task as completed."""
        # In production, this would update your database
        return {
            "success": True,
            "task_id": task_id,
            "message": f"Marked task {task_id} as completed"
        }

    @function_tool
    def update_task_tool(
        self,
        task_id: int,
        title: Optional[str] = None,
        description: Optional[str] = None
    ) -> dict:
        """Update an existing task."""
        # In production, this would update your database
        return {
            "success": True,
            "task_id": task_id,
            "updates": {"title": title, "description": description},
            "message": f"Updated task {task_id}"
        }

    @function_tool
    def delete_task_tool(self, task_id: int) -> dict:
        """Delete a task by ID."""
        # In production, this would delete from your database
        return {
            "success": True,
            "deleted_task_id": task_id,
            "message": f"Deleted task {task_id}"
        }

# Usage example
async def main():
    assistant = TodoAssistant(user_id="user123")

    # Setup MCP server if needed
    await assistant.setup_mcp_server()

    # Process a message
    result = await assistant.process_message("Add a task to buy groceries")
    print(result["response"])

    # Cleanup
    if assistant.mcp_server:
        await assistant.mcp_server.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

MCP Server Patterns

1. Basic MCP Server Implementation

# app/mcp/todo_server.py
import asyncio
import json
import sys
from typing import Dict, Any

class TodoMCPServer:
    """MCP server for todo operations."""

    def __init__(self):
        self.tools = {
            "add_todo": self.add_todo,
            "list_todos": self.list_todos,
            "complete_todo": self.complete_todo,
            "update_todo": self.update_todo,
            "delete_todo": self.delete_todo
        }

    async def run(self):
        """Run the MCP server loop."""
        while True:
            try:
                line = await self._read_line()
                if not line:
                    break

                request = json.loads(line)

                # Process the request
                response = await self._process_request(request)

                # Send response
                await self._write_response(response)

            except KeyboardInterrupt:
                break
            except Exception as e:
                error_response = {
                    "error": str(e),
                    "type": "error"
                }
                await self._write_response(error_response)

    async def _read_line(self) -> str:
        """Read a line from stdin."""
        return await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)

    async def _write_response(self, response: Dict[str, Any]):
        """Write response to stdout."""
        print(json.dumps(response), flush=True)

    async def _process_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Process an MCP request."""
        if request.get("method") == "tools/list":
            return {
                "result": {
                    "tools": [
                        {
                            "name": name,
                            "description": f"Perform {name} operation",
                            "inputSchema": {
                                "type": "object",
                                "properties": {},
                                "required": []
                            }
                        }
                        for name in self.tools.keys()
                    ]
                },
                "id": request.get("id")
            }

        elif request.get("method") == "tools/call":
            tool_name = request.get("params", {}).get("name")
            tool_args = request.get("params", {}).get("arguments", {})

            if tool_name in self.tools:
                result = await self.tools[tool_name](**tool_args)
                return {
                    "result": result,
                    "id": request.get("id")
                }
            else:
                return {
                    "error": {"message": f"Unknown tool: {tool_name}"},
                    "id": request.get("id")
                }

        return {
            "error": {"message": "Unsupported method"},
            "id": request.get("id")
        }

    async def add_todo(self, title: str, description: str = "") -> Dict[str, Any]:
        """Add a new todo."""
        return {
            "success": True,
            "todo_id": 123,
            "title": title,
            "description": description
        }

    async def list_todos(self, status: str = "all") -> Dict[str, Any]:
        """List todos with optional filtering."""
        return {
            "todos": [
                {"id": 1, "title": "Sample task", "completed": False},
                {"id": 2, "title": "Another task", "completed": True}
            ]
        }

    async def complete_todo(self, todo_id: int) -> Dict[str, Any]:
        """Complete a todo."""
        return {
            "success": True,
            "todo_id": todo_id,
            "status": "completed"
        }

    async def update_todo(self, todo_id: int, title: str = None, description: str = None) -> Dict[str, Any]:
        """Update a todo."""
        return {
            "success": True,
            "todo_id": todo_id,
            "updates": {"title": title, "description": description}
        }

    async def delete_todo(self, todo_id: int) -> Dict[str, Any]:
        """Delete a todo."""
        return {
            "success": True,
            "deleted_todo_id": todo_id
        }

if __name__ == "__main__":
    server = TodoMCPServer()
    asyncio.run(server.run())

Best Practices

1. Security Considerations

# Secure MCP server with validation
class SecureMCPServer:
    def __init__(self, allowed_tools: List[str]):
        self.allowed_tools = set(allowed_tools)
        self.tools = {
            "safe_operation": self.safe_operation,
            # Only safe operations
        }

    async def _validate_request(self, request: Dict[str, Any]) -> bool:
        """Validate incoming requests."""
        tool_name = request.get("params", {}).get("name")
        if tool_name not in self.allowed_tools:
            return False
        return True

2. Error Handling

# Robust error handling in tools
@function_tool
def robust_task_tool(
    task_id: int,
    title: Optional[str] = None
) -> dict:
    """Tool with comprehensive error handling."""
    try:
        if not title:
            raise ValueError("Title is required")

        if task_id <= 0:
            raise ValueError("Task ID must be positive")

        # Perform operation
        result = perform_operation(task_id, title)

        return {
            "success": True,
            "result": result
        }

    except ValueError as e:
        return {
            "success": False,
            "error": str(e),
            "error_type": "validation"
        }
    except Exception as e:
        return {
            "success": False,
            "error": "Internal server error",
            "error_type": "internal"
        }

Integration with Existing Systems

1. Connecting to Todo Services

# app/agents/todo_service_integration.py
from app.services.task_service import TaskService

class TodoAgentWithService:
    def __init__(self, user_id: str, task_service: TaskService):
        self.user_id = user_id
        self.task_service = task_service
        
        # Setup OpenRouter client using specific configuration
        openrouter_api_key = os.getenv("OPENROUTER_API_KEY")
        openrouter_base_url = os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")
        openrouter_model = os.getenv("OPENROUTER_MODEL", "mistralai/devstral-2512:free")
        
        if not openrouter_api_key:
            raise ValueError("OPENROUTER_API_KEY is not set in environment variables")
        
        external_client = AsyncOpenAI(
            api_key=openrouter_api_key,
            base_url=openrouter_base_url,
        )

        # Define the model to use (using available OpenRouter models)
        model = OpenAIChatCompletionsModel(
            model=openrouter_model,  # Using mistralai/devstral-2512:free as specified
            openai_client=external_client
        )

        # Create run configuration with OpenRouter
        self.config = RunConfig(
            model=model,
            model_provider=external_client,
            # Add other configuration as needed
        )

    @function_tool
    async def add_task_from_service(
        self,
        title: str,
        description: Optional[str] = None
    ) -> dict:
        """Add task using the actual task service."""
        try:
            task = await self.task_service.create_task(
                user_id=self.user_id,
                title=title,
                description=description
            )

            return {
                "success": True,
                "task_id": task.id,
                "title": task.title,
                "message": f"Created task: {task.title}"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

Example Usage Scenarios

Production Considerations

1. Resource Management: Properly manage MCP server lifecycle with connect/cleanup
2. Security: Validate all inputs and restrict tool access appropriately
3. Error Handling: Implement comprehensive error handling and recovery
4. Monitoring: Log agent runs and tool calls for debugging
5. Rate Limiting: Implement rate limiting for expensive operations
6. Caching: Cache frequently accessed data to improve performance
7. OpenRouter Integration: Use specific configuration with FREE model mistralai/devstral-2512:free

Production Standard: This skill ensures secure, efficient, and well-structured OpenAI Agents SDK implementations with proper OpenRouter integration, MCP server integration, context management, and production-ready error handling.