03_LangChain-MCP解析

LangChain-MCP

大模型和外部工具交互的标准化协议

模型上下文协议(MCP ModelContextProtocol)正迅速成为AI领域的核心基础设施标准，它通过标准化大语言模型与外部工具的交互方式，解决了AI应用开发中的关键瓶颈，推动了智能体从实验室走向商业化的进程

MCP的原理

MCP的核心原理：

与上一张图异曲同工：

MCP 的核心原理是将互联网服务（高德、谷歌）或本地操作系统 API（文件系统、数据库、终端）封装成 AI 智能体能够理解和使用的 Tools 工具，让 AI 智能体能够自由地调用这些 Tools 工具实现复杂的业务逻辑和功能。

基于MCP的智能体架构

案例-高德地图API

[概述-MCP Server

高德地图API](https://lbs.amap.com/api/mcp-server/summary)

import asyncio
import json
import os
from typing import List, Dict, Any
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from pydantic import BaseModel, Field, SecretStr
from langchain.agents import create_agent
# LangChain核心组件
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# MCP相关组件
from langchain_mcp_adapters.client import MultiServerMCPClient

# 文件管理工具
from langchain_community.agent_toolkits import FileManagementToolkit

# 其他工具
import aiofiles
import datetime

# 加载环境变量
load_dotenv()

print("正在导入依赖模块...")

def load_model_config(config_path="../config.json"):
    """从JSON配置文件加载模型配置"""
    # 获取当前脚本所在目录
    current_dir = os.path.dirname(os.path.abspath(__file__))
    # 构建配置文件的绝对路径
    config_file_path = os.path.join(current_dir, config_path)
    
    try:
        with open(config_file_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        
        # 获取默认模型配置
        default_model = config.get("default_model", "kimi")
        model_config = config["models"][default_model]
        
        return model_config
    except FileNotFoundError:
        raise FileNotFoundError(f"配置文件未找到: {config_file_path}")
    except KeyError as e:
        raise KeyError(f"配置文件格式错误，缺少必要字段: {e}")

def load_service_config(service_name: str, config_path="../config.json"):
    """从JSON配置文件加载服务配置"""
    # 获取当前脚本所在目录
    current_dir = os.path.dirname(os.path.abspath(__file__))
    # 构建配置文件的绝对路径
    config_file_path = os.path.join(current_dir, config_path)
    
    try:
        with open(config_file_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        
        # 获取服务配置
        service_config = config.get("services", {}).get(service_name, {})
        
        if not service_config:
            raise KeyError(f"配置文件中未找到服务 '{service_name}' 的配置")
            
        return service_config
    except FileNotFoundError:
        raise FileNotFoundError(f"配置文件未找到: {config_file_path}")
    except KeyError as e:
        raise KeyError(f"配置文件格式错误: {e}")

def load_file_management_config(config_path="../config.json"):
    """从JSON配置文件加载文件管理工具配置"""
    # 获取当前脚本所在目录
    current_dir = os.path.dirname(os.path.abspath(__file__))
    # 构建配置文件的绝对路径
    config_file_path = os.path.join(current_dir, config_path)
    
    try:
        with open(config_file_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        
        # 获取文件管理配置
        file_config = config.get("file_management", {})
        
        if not file_config:
            # 如果没有配置，则使用默认值
            print("没有找到文件管理配置...")

        return file_config
    except FileNotFoundError:
        raise FileNotFoundError(f"配置文件未找到: {config_file_path}")
    except KeyError as e:
        raise KeyError(f"配置文件格式错误: {e}")

# ========== MCP客户端实现 ==========
async def create_mcp_client(max_retries=2):
    """创建高德MCP客户端，带重试机制和模拟工具后备"""
    # 从配置文件读取高德地图配置
    try:
        amap_config = load_service_config("amap")
        amap_key = amap_config.get("api_key")

        if not amap_key or amap_key == "your_amap_key_here":
            print("⚠️  API密钥无效，使用模拟工具")
            return None, []

        amap_base_url = amap_config.get("base_url", "https://mcp.amap.com")

    except Exception as e:
        print(f"读取配置失败: {str(e)}，使用模拟工具")
        return None, []

    print("正在连接高德MCP服务...")

    # 尝试连接，带重试机制
    for attempt in range(max_retries):
        try:
            client = MultiServerMCPClient({
                "amap": {
                    "url": f"{amap_base_url}/sse?key={amap_key}",
                    "transport": "sse",
                }
            })

            # 获取可用工具
            tools = await client.get_tools()
            print(f"✅ 成功连接，已获取 {len(tools)} 个MCP工具")
            return client, tools

        except Exception as e:
            print(f"第 {attempt + 1} 次连接尝试失败: {type(e).__name__}: {str(e)[:100]}...")
            if attempt < max_retries - 1:
                print("等待2秒后重试...")
                await asyncio.sleep(2)
            else:
                print("❌ 连接高德MCP服务失败，切换到模拟工具模式")
                return None, []

    return None, []

# ========== 数据结构定义 ==========
class TravelPlan(BaseModel):
    destination: str = Field(description="目的地城市")
    duration_days: int = Field(description="旅行天数")
    weather_forecast: str = Field(description="天气预报信息")
    daily_schedule: List[Dict[str, Any]] = Field(description="每日行程安排")
    transportation_links: List[str] = Field(description="交通链接(打车、导航等)")
    map_links: List[str] = Field(description="地图相关链接")
    recommendations: List[str] = Field(description="旅行建议")

# ========== 本地工具定义（未使用）  ==========
@tool
def save_to_file(content: str, filename: str) -> str:
    """将内容保存到文件"""
    try:
        # 从配置文件获取输出目录配置
        try:
            file_config = load_file_management_config()
            output_dir = file_config.get("output_dir", "./output")
        except:
            # 如果配置读取失败，使用环境变量或默认值
            print("如果配置读取失败...")

        # 确保输出目录存在
        os.makedirs(output_dir, exist_ok=True)

        filepath = os.path.join(output_dir, filename)

        # 异步写入文件
        async def write_file():
            async with aiofiles.open(filepath, 'w', encoding='utf-8') as f:
                await f.write(content)

        # 在事件循环中运行
        if asyncio.get_event_loop().is_running():
            asyncio.create_task(write_file())
        else:
            asyncio.run(write_file())

        return f"文件已保存到: {filepath}"
    except Exception as e:
        return f"保存文件失败: {str(e)}"

# ========== 本地工具(模拟MCP工具)（未使用） ==========
@tool
def mock_poi_search(city: str, keywords: str = "旅游景点") -> str:
    """模拟POI搜索工具"""
    mock_results = {
        "杭州": [
            {"name": "西湖", "address": "杭州市西湖区", "type": "风景名胜"},
            {"name": "灵隐寺", "address": "杭州市西湖区", "type": "宗教场所"},
            {"name": "千岛湖", "address": "杭州市淳安县", "type": "自然景观"},
            {"name": "宋城", "address": "杭州市西湖区", "type": "文化娱乐"}
        ],
        "default": [
            {"name": "市中心广场", "address": "市中心", "type": "地标建筑"},
            {"name": "博物馆", "address": "文化区", "type": "文化场所"}
        ]
    }
    results = mock_results.get(city, mock_results["default"])
    return json.dumps(results, ensure_ascii=False)

@tool
def mock_route_planning(start: str, end: str, mode: str = "driving") -> str:
    """模拟路线规划工具"""
    return f"从{start}到{end}的{mode}路线规划已完成，预计耗时30分钟，距离15公里"

@tool
def mock_weather_query(city: str, date: str) -> str:
    """模拟天气查询工具"""
    return f"{city}在{date}的天气：晴天，气温25-30°C，适合户外活动"

@tool
def mock_geocode(address: str) -> str:
    """模拟地理编码工具"""
    return f"{address}的坐标：经度120.1551，纬度30.2741"

# ========== 主要功能实现 ==========
async def create_travel_agent():
    """创建集成MCP工具的旅行规划智能体"""
    print("正在创建旅行规划智能体...")

    # 1. 获取MCP工具(高德地图)
    client, mcp_tools = await create_mcp_client()

    # 如果MCP连接失败，使用模拟工具
    if not mcp_tools:
        print("⚠️ MCP连接失败...")


    # 2. 获取文件管理工具配置
    try:
        file_config = load_file_management_config()
        root_dir = file_config.get("root_dir", "./temp")
        selected_tools = file_config.get("selected_tools", ["write_file", "read_file", "list_directory"])

    except Exception as e:
        print(f"读取文件管理配置失败，使用默认配置: {str(e)}")

    # 3. 获取文件管理工具(内置工具)
    file_toolkit = FileManagementToolkit(
        root_dir=root_dir,
        selected_tools=selected_tools
    )
    file_tools = file_toolkit.get_tools()

    # 4. 合并所有工具
    all_tools = mcp_tools + file_tools + [save_to_file]
    print(f"总共可用工具数量: {len(all_tools)}")

    # 5. 加载LLM配置
    model_config = load_model_config()
    llm = ChatOpenAI(
        model=model_config["model"],
        base_url=model_config["base_url"],
        api_key=SecretStr(model_config["api_key"]),
        temperature=model_config["temperature"],
    )

    # 6. 创建智能体
    agent = create_agent(
        model=llm,
        tools=all_tools,
        system_prompt=SYSTEM_PROMPT # 这是agent的默认系统提示
    )

    return agent, client

# ========== 提示词模板 ==========
# 定义系统提示（不变的部分）
SYSTEM_PROMPT = """你是一个专业的旅行规划助手，现在必须立即使用高德MCP工具为用户制定详细的旅行计划。

强制要求：
1. 必须主动调用工具，不允许只提供文字说明
2. 必须按顺序执行以下工具调用：
   - POI搜索：查询目的地热门景点
   - 地理编码：获取景点精确坐标
   - 路线规划：设计每日行程路线
   - 天气查询：获取出行期间天气
   - 生成导航链接和地图链接
   - 保存为HTML文件
3. 每次工具调用后都要分析结果并继续下一步
4. 最终输出完整的结构化旅行计划

要求:
- 考虑天气状况和出行时间
- 提供具体的景区位置和路线规划
- 生成可以直接使用的导航链接
- 最终结果要结构化输出并保存到文件"""


# ========== 主函数 ==========
async def main():
    """主函数：执行完整的旅行规划流程"""
    print("=" * 60)
    print("开始执行高德MCP旅行规划系统")
    print("=" * 60)

    try:
        # 1. 创建智能体
        agent, client = await create_travel_agent()

        # 2. 准备用户输入 - 更加强制性的指令
        user_input = """现在立即执行以下操作，必须调用工具：

操作1: 调用POI搜索工具，查询杭州的热门旅游景点
操作2: 调用地理编码工具，获取西湖的精确坐标
操作3: 调用路线规划工具，规划从西湖到灵隐寺的路线
操作4: 调用天气查询工具，查询杭州未来一周的天气

每完成一个操作后继续下一个，不要等待确认。
最终生成包含所有信息的旅行计划。"""

        print("\n正在处理请求...")

        # 4. 使用提示词模板格式化输入，然后执行智能体
        prompt = ChatPromptTemplate.from_messages([
            ("system", SYSTEM_PROMPT),  # 系统指令
            ("human", "{input}"),  # 用户输入占位符
        ])
        formatted_input = prompt.format(input=user_input) # 动态注入用户输入
        print(f"prompt: {formatted_input}")
        response = await agent.ainvoke({"input": formatted_input})# 传递格式化后的完整

        # 调试：打印响应结构
        print(f"\n响应类型: {type(response)}")
        print(f"响应键值: {list(response.keys()) if isinstance(response, dict) else 'Not a dict'}")

        # 详细查看消息内容和工具调用
        if isinstance(response, dict) and "messages" in response:
            messages = response["messages"]
            print(f"\n消息数量: {len(messages)}")
            for i, msg in enumerate(messages):
                print(f"\n--- 消息 {i+1} ---")
                print(f"类型: {type(msg).__name__}")
                if hasattr(msg, 'content'):
                    print(f"内容长度: {len(str(msg.content))} 字符")
                    print(f"内容预览: {str(msg.content)[:200]}...")
                if hasattr(msg, 'tool_calls') and msg.tool_calls:
                    print(f"工具调用: {msg.tool_calls}")
                if hasattr(msg, 'additional_kwargs'):
                    print(f"额外信息: {msg.additional_kwargs}")

        print(f"\n完整响应: {response}")

        # 获取最终消息
        if isinstance(response, dict) and "messages" in response:
            last_msg = response["messages"][-1]
            print(f"\n最终内容：{repr(last_msg.content)}")

            # 检查是否有工具调用
            if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
                print(f"✅ 检测到工具调用: {len(last_msg.tool_calls)} 个")
                for tool_call in last_msg.tool_calls:
                    print(f"  - 工具: {tool_call.get('name', 'unknown')}")
            else:
                print("⚠️  未检测到工具调用")
        # 5. 输出结果
        print("\n" + "=" * 60)
        print("旅行规划完成!")
        print("=" * 60)


    except Exception as e:
        print(f"\n执行过程中出现错误: {str(e)}")
        import traceback
        traceback.print_exc()

    finally:
        print("\n程序执行完毕")

# ========== 运行入口 ==========
if __name__ == "__main__":
    print("启动基于高德MCP的复杂路径规划 + 可视化展示...")
    asyncio.run(main())

效果