AI 智能体 - 模型上下文协议

pip install fastmcp

from fastmcp import FastMCP, Contextimport subprocessimport os
# 初始化 Server# dependencies 参数允许你声明该 Server 运行所需的依赖，FastMCP 会自动管理环境mcp = FastMCP("DevOps Server", dependencies=["psutil"])
# --- 1. 定义资源 (Resources) ---# 资源通常是只读的数据流@mcp.resource("system://logs/{service_name}")def get_service_logs(service_name: str) -> str:    """获取指定服务的最近 50 行日志"""    log_path = f"/var/log/{service_name}.log"    if not os.path.exists(log_path):        return "Log file not found."        # 模拟读取日志    try:        # 在真实场景中，这里可能是 tail -n 50        with open(log_path, "r") as f:            lines = f.readlines()[-50:]            return "".join(lines)    except Exception as e:        return f"Error reading logs: {str(e)}"
# --- 2. 定义工具 (Tools) ---# 工具是可以改变系统状态的操作@mcp.tool()def restart_service(service_name: str, force: bool = False) -> str:    """    重启指定的系统服务。        Args:        service_name: 服务名称 (如 nginx, postgresql)        force: 是否强制重启 (kill -9)    """    # 简单的安全检查（在真实 MCP 中，可以在 Client 端做鉴权）    allowed_services = ["nginx", "app_worker"]    if service_name not in allowed_services:        return f"Error: Service '{service_name}' is not allowed to be managed via MCP."
    cmd = ["systemctl", "restart", service_name]    if force:        # 仅作演示，实际生产中需谨慎        pass             try:        # 执行命令        result = subprocess.run(cmd, capture_output=True, text=True)        if result.returncode == 0:            return f"Service '{service_name}' restarted successfully."        else:            return f"Failed to restart: {result.stderr}"    except Exception as e:        return f"Execution error: {str(e)}"
# --- 3. 定义提示词 (Prompts) ---# 预定义的交互模板，帮助 LLM 更好地使用这些工具@mcp.prompt("diagnose_error")def diagnose_prompt(service_name: str) -> str:    """生成用于诊断服务错误的 Prompt"""    return f"""    请作为一名资深运维工程师，分析以下服务的状态。        1. 首先，使用 `get_service_logs` 读取资源 `xxx://logs/{service_name}`。    2. 分析日志中的 ERROR 或 WARNING 信息。    3. 如果认为是临时故障，调用 `restart_service` 工具。    4. 如果是配置错误，请给出修改建议。    """
if __name__ == "__main__":    # 启动服务器（默认使用 Stdio 模式，适合本地 CLI 集成）    mcp.run()

# 在终端运行fastmcp inspect my_server.py

pip install mcp langchain-openai langgraph

import asynciofrom contextlib import AsyncExitStackfrom typing import List
from langchain_core.tools import StructuredToolfrom langchain_openai import ChatOpenAIfrom langgraph.prebuilt import create_react_agent
# MCP 官方 SDKfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_client
class LangChainMCPClient:    """    一个通用的适配器，将 MCP Server 转换为 LangChain 可用的组件。    """    def __init__(self, command: str, args: List[str]):        self.params = StdioServerParameters(command=command, args=args)        self.session: ClientSession | None = None        self.exit_stack = AsyncExitStack()
    async def connect(self):        """建立与 MCP Server 的连接"""        # 启动子进程并建立通信通道        read, write = await self.exit_stack.enter_async_context(stdio_client(self.params))        self.session = await self.exit_stack.enter_async_context(            ClientSession(read, write)        )        await self.session.initialize()                # 发现工具        tools_list = await self.session.list_tools()        print(f"🔗 Connected to MCP Server. Found {len(tools_list.tools)} tools.")        return tools_list
    async def get_langchain_tools(self) -> List[StructuredTool]:        """将 MCP Tools 转换为 LangChain Tools"""        if not self.session:            await self.connect()                    mcp_tools = (await self.session.list_tools()).tools        langchain_tools = []
        for tool in mcp_tools:            # 闭包捕获 tool.name            async def _tool_wrapper(                tool_name=tool.name,                 **kwargs            ):                # 实际调用 MCP Server                result = await self.session.call_tool(tool_name, arguments=kwargs)                # MCP 返回的是 Content 列表，通常我们要提取文本                return result.content[0].text
            # 构建 LangChain Tool            lc_tool = StructuredTool.from_function(                func=None,                coroutine=_tool_wrapper, # 使用异步调用                name=tool.name,                description=tool.description,                # MCP 的 inputSchema 是 JSON Schema，LangChain 可以直接用（或稍作转换）                # 这里为了简化，我们依赖 LangChain 的自动推断，或者你需要手动转换 Pydantic            )            langchain_tools.append(lc_tool)                    return langchain_tools
    async def cleanup(self):        await self.exit_stack.aclose()
# --- 业务逻辑 ---
async def main():    # 假设我们有一个运行 SQLite 的 MCP Server    # (你可以使用官方的 @modelcontextprotocol/server-sqlite)    # 这里我们连接到一个本地运行的 Python MCP Server    client = LangChainMCPClient(        command="python",         args=["my_devops_server.py"] # 上一节写的 Server    )
    try:        # 1. 获取工具        tools = await client.get_langchain_tools()                # 2. 初始化 LLM        llm = ChatOpenAI(model="gpt-4o", temperature=0)                # 3. 构建 LangGraph Agent (ReAct 模式)        agent_executor = create_react_agent(llm, tools)                # 4. 执行任务        print("\n🤖 Agent 启动...")        query = "请检查 nginx 服务的日志，如果发现错误，请帮我重启它。"                # Stream 模式查看执行过程        async for event in agent_executor.astream(            {"messages": [("user", query)]}        ):            for value in event.values():                print("--- Step ---")                print(value["messages"][-1].content)                    finally:        await client.cleanup()
if __name__ == "__main__":    asyncio.run(main())

import osfrom google.adk.agents import LlmAgentfrom google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
# 1. 准备目标目录TARGET_DIR = os.path.abspath("./workspace")os.makedirs(TARGET_DIR, exist_ok=True)
# 2. 定义智能体agent = LlmAgent(    model='gemini-2.0-flash',    name='FileAssistant',    instruction=f'你是一个文件管理助手。你可以读写 {TARGET_DIR} 目录下的文件。',    # 3. 注入 MCP 工具集    tools=[        MCPToolset(            connection_params=StdioServerParameters(                # 使用 npx 运行社区提供的 MCP Server                command='npx',                args=[                    "-y",                     "@modelcontextprotocol/server-filesystem",                     TARGET_DIR                ],            ),            # 可选：权限控制，只暴露读接口，隐藏写接口            # tool_filter=['list_directory', 'read_file']        )    ])
# 注意：ADK 会自动处理 Server 的启动、保活和关闭

from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, HttpServerParameters
remote_tools = MCPToolset(    connection_params=HttpServerParameters(        url="http://mcp-database-service.internal:8000/sse"    ))

# 伪代码示例@mcp.tool()async def consult_legal_department(query: str) -> str:    """咨询法律部门（这是一个子智能体）"""    # 内部调用 LangGraph Agent    response = await legal_agent.ainvoke({"messages": [query]})    return response["messages"][-1].content