LangChain 接入 MCP 实战指南：解锁模型上下文管理新范式

pip install langchain langchain-community mcp openai

# simple_mcp_server.py
import asyncio
from typing import List, Optional
from mcp import Server, ClientSession
from mcp.shared.models import ContextItem, ListContextItemsResult

class SimpleContextManager:
    """一个简单的内存上下文管理器示例"""
    def __init__(self):
        self.contexts: List[ContextItem] = []

    async def list_items(self) -> ListContextItemsResult:
        """列出所有上下文项"""
        return ListContextItemsResult(items=self.contexts)

    async def add_item(self, item: ContextItem):
        """添加上下文项（这里实现简单的去重）"""
        # 简单示例：如果存在同名的项，则替换内容
        for i, ctx in enumerate(self.contexts):
            if ctx.name == item.name:
                self.contexts[i] = item
                return
        self.contexts.append(item)

    async def get_relevant_context(self, query: str, limit: int = 2) -> List[ContextItem]:
        """根据查询返回最相关的上下文（这里实现简单的关键词匹配）"""
        # 这是一个非常简化的“检索”逻辑。生产环境应使用向量数据库。
        relevant = []
        query_lower = query.lower()
        for ctx in self.contexts:
            if query_lower in ctx.content.lower():
                relevant.append(ctx)
                if len(relevant) >= limit:
                    break
        return relevant

# 创建 MCP 服务器实例
server = Server()
context_manager = SimpleContextManager()

@server.list_context_items()
async def handle_list_items(session: ClientSession) -> ListContextItemsResult:
    return await context_manager.list_items()

@server.add_context_item()
async def handle_add_item(session: ClientSession, item: ContextItem) -> None:
    await context_manager.add_item(item)

@server.get_context()
async def handle_get_context(session: ClientSession, query: str) -> List[ContextItem]:
    # 在实际协议中，可能需要更复杂的请求体，这里做了简化
    return await context_manager.get_relevant_context(query)

async def main():
    # 启动服务器，监听本地端口 8000
    async with await server.run_tcp_server("localhost", 8000) as server_handle:
        print("MCP 服务器已在 localhost:8000 启动")
        await server_handle.serve_forever()

if __name__ == "__main__":
    asyncio.run(main())

# langchain_mcp_client.py
import asyncio
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.tools.mcp import MCPClient, create_mcp_tool

async def main():
    # 1. 初始化 MCP 客户端，连接到我们的服务器
    # 注意：`langchain-community` 中的 MCP 集成可能仍在演进，以下为示例性代码。
    # 实际使用时请查阅最新文档。
    async with MCPClient(server_url="http://localhost:8000") as client:
        # 2. 基于 MCP 客户端的能力创建 LangChain Tool
        # 假设我们的服务器提供了 `add_context` 和 `query_context` 工具
        tools = []
        
        # 创建“添加上下文”的工具
        add_context_tool = create_mcp_tool(
            client,
            tool_name="add_knowledge",
            description="添加一段知识或信息到上下文库中。输入应为要存储的文本内容。"
        )
        tools.append(add_context_tool)
        
        # 创建“查询上下文”的工具
        query_context_tool = create_mcp_tool(
            client,
            tool_name="search_knowledge",
            description="从上下文库中搜索与问题相关的知识。输入应为你的问题。"
        )
        tools.append(query_context_tool)
        
        # 3. 初始化 LLM
        llm = ChatOpenAI(model="gpt-4o", temperature=0)
        
        # 4. 创建提示词模板
        prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个拥有外部知识库的助手。你可以使用工具来存储或查找信息。在回答用户问题时，请先尝试从知识库中搜索相关信息。"),
            ("placeholder", "{chat_history}"),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ])
        
        # 5. 创建 Agent 和执行器
        agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
        agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
        
        # 6. 运行示例对话
        print("=== 示例 1：存储信息 ===")
        result1 = await agent_executor.ainvoke({
            "input": "请记住，我们公司的产品‘AI助手’的核心优势是实时翻译和代码生成。",
            "chat_history": []
        })
        print(f"结果: {result1['output']}\n")
        
        print("=== 示例 2：基于存储的信息回答问题 ===")
        result2 = await agent_executor.ainvoke({
            "input": "你们公司的‘AI助手’有什么特点？",
            "chat_history": [] # 在实际多轮对话中，这里应包含历史消息
        })
        print(f"结果: {result2['output']}")

if __name__ == "__main__":
    asyncio.run(main())