The McpAgent extends ToolAgent with the ability to connect to MCP (Model Context Protocol) servers. It provides access to tools, resources, and prompts from external servers, enabling rich integrations with external systems and data sources.McpAgent uses an MCPAggregator to manage multiple server connections and provides a unified interface for accessing capabilities across all connected servers.
# List all tools from all serverstools_result = await agent.list_tools()for tool in tools_result.tools: print(f"Tool: {tool.name}") print(f"Description: {tool.description}") print(f"Schema: {tool.inputSchema}")
from mcp.types import CallToolResult# Call a specific MCP toolresult: CallToolResult = await agent.call_tool( name="fetch__fetch_url", # Namespaced tool name arguments={"url": "https://example.com"},)if not result.isError: print(result.content[0].text)else: print(f"Error: {result.content[0].text}")
# List all resources from all serversresources = await agent.list_resources()for server_name, resource_list in resources.items(): print(f"Server: {server_name}") for resource_uri in resource_list: print(f" - {resource_uri}")
# Attach resource to a messageresponse = await agent.with_resource( prompt_content="Analyze this file", resource_uri="file:///data/report.csv", server_name="filesystem")print(response)# Or get embedded resources for custom useembedded = await agent.get_embedded_resources( resource_uri="file:///config.json")
# Apply a prompt template from an MCP serverresponse = await agent.apply_prompt( prompt="code_review", arguments={"language": "python", "file": "agent.py"}, namespace="dev_tools" # Server name)print(response)# Use as persistent templateawait agent.apply_prompt( prompt="system_context", as_template=True # Always included in context)
# Get list of attached serversattached = agent.list_attached_mcp_servers()print(f"Connected servers: {attached}")# Get detailed server statusstatus = await agent.get_server_status()for server_name, server_status in status.items(): print(f"{server_name}: {server_status.state}") if server_status.error: print(f" Error: {server_status.error}")
# The agent automatically gets bash/shell tools when enabledresult = await agent.send( "List all Python files in the current directory")# Agent can now execute: ls *.py
# Use server instructions in system promptconfig = AgentConfig( name="template_agent", instruction="""You are a helpful assistant. {{serverInstructions}} {{agentSkills}} """, servers=["fetch"], skills=["research"])# Templates are resolved during initializationagent = McpAgent(config, context=core.context)await agent.attach_llm(ModelFactory.create_factory("gpt-4o-mini"))await agent.initialize()# View resolved instructionprint(agent.instruction)
config = AgentConfig( name="researcher", instruction="You are a research assistant that can search the web and read files.", servers=["fetch", "filesystem"], tools={ "fetch": ["fetch_url", "search_*"], "filesystem": ["read_*"] }, model="gpt-4o")
config = AgentConfig( name="analyst", instruction="You analyze data from various sources.", servers=["postgres", "filesystem"], model="claude-3-5-sonnet-20241022")
config = AgentConfig( name="dev_assistant", instruction="You help with software development tasks.", servers=["filesystem"], skills=["code-review", "testing"], shell=True, model="gpt-4o")