68c800d2a3
### Summary: This enables users to **use** MCP inside the SDK. 1. You add a list of MCP servers to `Agent`, via `mcp_server=[...]` 2. When an agent runs, we look up its MCP tools and add them to the list of tools. 3. When a tool call occurs, we call the relevant MCP server. Notes: 1. There's some refactoring to make sure we send the full list of tools to the Runner/Model etc. 2. Right now, you could have a locally defined tool that conflicts with an MCP defined tool. I didn't add errors for that, will do in a followup. ### Test Plan: See unit tests. Also has an end to end example next PR.
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
import json
|
|
import shutil
|
|
from typing import Any
|
|
|
|
from mcp import Tool as MCPTool
|
|
from mcp.types import CallToolResult, TextContent
|
|
|
|
from agents.mcp import MCPServer
|
|
|
|
tee = shutil.which("tee") or ""
|
|
assert tee, "tee not found"
|
|
|
|
|
|
# Added dummy stream classes for patching stdio_client to avoid real I/O during tests
|
|
class DummyStream:
|
|
async def send(self, msg):
|
|
pass
|
|
|
|
async def receive(self):
|
|
raise Exception("Dummy receive not implemented")
|
|
|
|
|
|
class DummyStreamsContextManager:
|
|
async def __aenter__(self):
|
|
return (DummyStream(), DummyStream())
|
|
|
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
pass
|
|
|
|
|
|
class FakeMCPServer(MCPServer):
|
|
def __init__(self, tools: list[MCPTool] | None = None):
|
|
self.tools: list[MCPTool] = tools or []
|
|
self.tool_calls: list[str] = []
|
|
self.tool_results: list[str] = []
|
|
|
|
def add_tool(self, name: str, input_schema: dict[str, Any]):
|
|
self.tools.append(MCPTool(name=name, inputSchema=input_schema))
|
|
|
|
async def connect(self):
|
|
pass
|
|
|
|
async def cleanup(self):
|
|
pass
|
|
|
|
async def list_tools(self):
|
|
return self.tools
|
|
|
|
async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
|
|
self.tool_calls.append(tool_name)
|
|
self.tool_results.append(f"result_{tool_name}_{json.dumps(arguments)}")
|
|
return CallToolResult(
|
|
content=[TextContent(text=self.tool_results[-1], type="text")],
|
|
)
|