在構建多Agent系統時,工具隔離是一個關鍵需求。通過Starlette和FastMCP,我們可以創建一個支持多路由隔離的工具服務器,讓不同Agent調用各自專屬的工具集。
核心思路:路由隔離工具集
from starlette.applications import Starlette
from starlette.routing import Mount
from mcp.server.fastmcp import FastMCP
# 創建多個獨立的MCP實例
github_mcp = FastMCP("GitHub API")
browser_mcp = FastMCP("Browser")
curl_mcp = FastMCP("Curl")
search_mcp = FastMCP("Search")
# 應用配置
app = Starlette(
routes=[
Mount("/github", app=github_mcp.sse_app()),
Mount("/browser", app=browser_mcp.sse_app()),
Mount("/curl", app=curl_mcp.sse_app()),
Mount("/search", app=search_mcp.sse_app()),
]
)
這種架構的關鍵優勢在於:
- 工具隔離:每個路由下的工具相互獨立
- 模塊化設計:按功能劃分工具組
- 獨立擴展:不同工具集可以單獨擴展和部署
- 權限控制:為不同Agent分配不同路由權限
工具定義示例
GitHub工具集(/github路由)
@github_mcp.tool()
async def search_github_repos(query: str, limit: int = 10) -> Dict[str, Any]:
"""搜索GitHub倉庫"""
url = f"https://api.github.com/search/repositories?q={query}&per_page={limit}"
response = requests.get(url)
return response.json()
瀏覽器工具集(/browser路由)
@browser_mcp.tool()
async def get_webpage_title(url: str) -> Dict[str, Any]:
"""獲取網頁標題"""
try:
response = requests.get(url, timeout=5)
# 簡單提取標題
import re
title_match = re.search(r'<title>(.*?)</title>', response.text, re.IGNORECASE | re.DOTALL)
title = title_match.group(1) if title_match else "未找到標題"
return {"url": url, "title": title}
except Exception as e:
return {"url": url, "error": str(e)}
客户端調用示例
不同Agent可以訪問各自的路由:
# GitHub Agent
github_client = Client("http://localhost:8000/github/sse")
await github_client.call_tool("search_github_repos", {"query": "AI"})
# 瀏覽器Agent
browser_client = Client("http://localhost:8000/browser/sse")
await browser_client.call_tool("get_webpage_title", {"url": "https://example.com"})
架構優勢分析
- 職責分離:每個工具組專注於特定領域
- 安全隔離:敏感工具(如curl)可單獨保護
- 性能優化:高負載工具可獨立擴展
- 版本管理:不同工具組可獨立升級
實際應用場景
- 多Agent協作系統:不同Agent使用不同工具集
- 微服務架構:每個路由作為獨立微服務
- 權限分級系統:根據用户角色分配路由訪問權限
- A/B測試環境:不同路由部署不同版本工具
總結
通過Starlette的路由掛載功能和FastMCP的工具管理能力,我們可以構建高度模塊化的Agent工具服務器。這種架構不僅實現了工具隔離,還為系統的可擴展性和安全性提供了堅實基礎。
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
啓動服務器後,不同Agent就可以通過各自的路由訪問專屬工具集,實現安全、高效的協作。
mcp 服務
from starlette.applications import Starlette
from starlette.routing import Mount
from mcp.server.fastmcp import FastMCP
import requests
from typing import Dict, Any
# Create multiple MCP servers
github_mcp = FastMCP("GitHub API")
browser_mcp = FastMCP("Browser")
curl_mcp = FastMCP("Curl")
search_mcp = FastMCP("Search")
@github_mcp.tool()
async def search_github_repos(query: str, limit: int = 10) -> Dict[str, Any]:
"""
搜索GitHub倉庫
Args:
query: 搜索關鍵詞
limit: 返回結果數量限制,默認10個
Returns:
包含倉庫信息的字典
"""
url = f"https://api.github.com/search/repositories?q={query}&per_page={limit}"
response = requests.get(url)
return response.json()
@browser_mcp.tool()
async def get_webpage_title(url: str) -> Dict[str, Any]:
"""
獲取網頁標題
Args:
url: 網頁地址
Returns:
包含網頁標題的字典
"""
try:
response = requests.get(url, timeout=5)
# 簡單提取標題
import re
title_match = re.search(r'<title>(.*?)</title>', response.text, re.IGNORECASE | re.DOTALL)
title = title_match.group(1) if title_match else "未找到標題"
return {"url": url, "title": title}
except Exception as e:
return {"url": url, "error": str(e)}
@curl_mcp.tool()
async def make_http_request(url: str, method: str = "GET", headers: Dict[str, str] = None) -> Dict[str, Any]:
"""
發起HTTP請求
Args:
url: 請求地址
method: HTTP方法,默認GET
headers: 請求頭
Returns:
HTTP響應結果
"""
try:
if headers is None:
headers = {}
response = requests.request(method, url, headers=headers)
return {
"status_code": response.status_code,
"headers": dict(response.headers),
"content": response.text[:1000] # 限制返回內容長度
}
except Exception as e:
return {"error": str(e)}
@search_mcp.tool()
async def search_text(query: str, texts: list) -> Dict[str, Any]:
"""
在文本列表中搜索關鍵詞
Args:
query: 搜索關鍵詞
texts: 要搜索的文本列表
Returns:
包含匹配文本的列表
"""
results = []
for i, text in enumerate(texts):
if query.lower() in text.lower():
results.append({"index": i, "text": text})
return {"query": query, "results": results}
app = Starlette(
routes=[
# Using settings-based configuration
Mount("/github", app=github_mcp.sse_app()),
Mount("/browser", app=browser_mcp.sse_app()),
Mount("/curl", app=curl_mcp.sse_app()),
Mount("/search", app=search_mcp.sse_app()),
]
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
客户請求
import asyncio
from fastmcp import Client
client = Client("http://localhost:8000/search/sse")
async def call_tool(name: str):
async with client:
result = await client.list_tools()
print(result)
result = await client.call_tool("search_text", {"query": name,"texts":["dfdsf","sdfsdfsf"]})
print(result)
asyncio.run(call_tool("Ford"))
client = Client("http://localhost:8000/curl/sse")
async def call_tool(name: str):
async with client:
result = await client.list_tools()
print(result)
asyncio.run(call_tool("Ford"))
client = Client("http://localhost:8000/browser/sse")
async def call_tool(name: str):
async with client:
result = await client.list_tools()
print(result)
asyncio.run(call_tool("Ford"))
client = Client("http://localhost:8000/github/sse")
async def call_tool(name: str):
async with client:
result = await client.list_tools()
print(result)
asyncio.run(call_tool("Ford"))
本文章為轉載內容,我們尊重原作者對文章享有的著作權。如有內容錯誤或侵權問題,歡迎原作者聯繫我們進行內容更正或刪除文章。