Richard Tang
120 lines
3.2 KiB
Python
120 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Example: Integrating MCP Servers with the Core Framework
|
|
|
|
This example demonstrates how to:
|
|
1. Register MCP servers programmatically
|
|
2. Use MCP tools in agents
|
|
3. Load MCP servers from configuration files
|
|
"""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
from framework.runner.runner import AgentRunner
|
|
|
|
|
|
async def example_1_programmatic_registration():
|
|
"""Example 1: Register MCP server programmatically"""
|
|
print("\n=== Example 1: Programmatic MCP Server Registration ===\n")
|
|
|
|
# Load an existing agent
|
|
runner = AgentRunner.load("exports/task-planner")
|
|
|
|
# Register tools MCP server via STDIO
|
|
num_tools = runner.register_mcp_server(
|
|
name="tools",
|
|
transport="stdio",
|
|
command="python",
|
|
args=["-m", "aden_tools.mcp_server", "--stdio"],
|
|
cwd="../tools",
|
|
)
|
|
|
|
print(f"Registered {num_tools} tools from tools MCP server")
|
|
|
|
# List all available tools
|
|
tools = runner._tool_registry.get_tools()
|
|
print(f"\nAvailable tools: {list(tools.keys())}")
|
|
|
|
# Run the agent with MCP tools available
|
|
result = await runner.run(
|
|
{"objective": "Search for 'Claude AI' and summarize the top 3 results"}
|
|
)
|
|
|
|
print(f"\nAgent result: {result}")
|
|
|
|
# Cleanup
|
|
runner.cleanup()
|
|
|
|
|
|
async def example_2_http_transport():
|
|
"""Example 2: Connect to MCP server via HTTP"""
|
|
print("\n=== Example 2: HTTP MCP Server Connection ===\n")
|
|
|
|
# First, start the tools MCP server in HTTP mode:
|
|
# cd tools && python mcp_server.py --port 4001
|
|
|
|
runner = AgentRunner.load("exports/task-planner")
|
|
|
|
# Register tools via HTTP
|
|
num_tools = runner.register_mcp_server(
|
|
name="tools-http",
|
|
transport="http",
|
|
url="http://localhost:4001",
|
|
)
|
|
|
|
print(f"Registered {num_tools} tools from HTTP MCP server")
|
|
|
|
# Cleanup
|
|
runner.cleanup()
|
|
|
|
|
|
async def example_3_config_file():
|
|
"""Example 3: Load MCP servers from configuration file"""
|
|
print("\n=== Example 3: Load from Configuration File ===\n")
|
|
|
|
# Create a test agent folder with mcp_servers.json
|
|
test_agent_path = Path("exports/task-planner")
|
|
|
|
# Copy example config (in practice, you'd place this in your agent folder)
|
|
import shutil
|
|
|
|
shutil.copy("examples/mcp_servers.json", test_agent_path / "mcp_servers.json")
|
|
|
|
# Load agent - MCP servers will be auto-discovered
|
|
runner = AgentRunner.load(test_agent_path)
|
|
|
|
# Tools are automatically available
|
|
tools = runner._tool_registry.get_tools()
|
|
print(f"Available tools: {list(tools.keys())}")
|
|
|
|
# Cleanup
|
|
runner.cleanup()
|
|
|
|
# Clean up the test config
|
|
(test_agent_path / "mcp_servers.json").unlink()
|
|
|
|
|
|
async def main():
|
|
"""Run all examples"""
|
|
print("=" * 60)
|
|
print("MCP Integration Examples")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
# Run examples
|
|
await example_1_programmatic_registration()
|
|
# await example_2_http_transport() # Requires HTTP server running
|
|
# await example_3_config_file()
|
|
# await example_4_custom_agent_with_mcp_tools()
|
|
|
|
except Exception as e:
|
|
print(f"\nError running example: {e}")
|
|
import traceback
|
|
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|