简介

大家好啊，今天给大家带来使用SKILL+MCP实现旅行规划Agent。先介绍下这两个概念：

• Skill（技能）：AI 的“行为模式”。它告诉 AI 在特定场景下如何思考和如何行动。

• MCP（Model Context Protocol）：AI 的“工具箱”。它告诉 AI 在特定场景下能使用什么工具。

核心区别对比

项目的目录结构如下：

python库依赖

openai>=1.0.0
mcp>=1.0.0
httpx
pydantic
asyncio

启动类

import asyncio
from mcp import StdioServerParameters
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client
from mcp_servers.travel_tools import server as travel_server
from llm_client import LLMClient
import os


class LLMSkillsMCPDemo:
    """LLM + Skills + MCP 演示"""

    def __init__(self):
        self.llm_client = LLMClient()

    async def run(self, user_request: str):
        """执行完整流程：用户请求 → LLM 决策 → MCP 工具调用 → 生成报告"""

        print("🚀 开始智能旅行规划...")
        print(f"📝 用户需求: {user_request}")

        # 1. 启动 MCP 服务器
        # server_params = StdioServerParameters(
        #     command="python",
        #     args=["-m", "from mcp_servers.travel_tools import server; server.run()"]
        # )
        server_params = StdioServerParameters(
            command="python",
            args=["run_server.py"]
        )

        # async with create_session(server_params) as session:
        async with stdio_client(server_params) as (read_stream, write_stream):
            async with ClientSession(read_stream, write_stream) as session:
                await session.initialize()
            # # 2. 初始化 MCP 连接
            # await session.initialize()

            # 3. 获取可用工具
                tools_result = await session.list_tools()
                available_tools = [{"name": tool.name, "description": tool.description} for tool in tools_result.tools]
                print(f"🛠️ 可用工具: {[tool['name'] for tool in available_tools]}")

                # 4. LLM 智能规划（核心决策）
                print("🧠 LLM 正在分析需求并制定计划...")
                travel_plan = await self.llm_client.plan_travel(user_request, available_tools)

                # 5. 解析 LLM 计划并调用相应工具
                print("🔧 执行计划并获取实时信息...")
                final_report = await self._execute_plan(travel_plan, session)

                return final_report

    async def _execute_plan(self, plan: str, session) -> str:
        """执行 LLM 生成的计划并调用 MCP 工具"""

        # 简化版：直接使用 LLM 生成的计划
        # 实际项目中可解析 plan 并动态调用工具
        return f"""
🌟 智能旅行规划完成！

以下是基于你的需求生成的个性化旅行计划：

{plan}

---
📊 信息说明：
- 航班、酒店、天气信息均为实时模拟数据
- 行程安排已考虑时间合理性和地理位置
- 预算估算包含交通、住宿、门票等主要费用
"""


async def main():
    """主函数"""
    demo = LLMSkillsMCPDemo()

    # 模拟用户请求
    user_request = """
    我想规划一个3天2晚的上海之旅，时间定在2月15日-17日。
    预算大概3000元左右，希望住在市中心交通方便的地方。
    我对文化历史景点比较感兴趣，还想体验当地美食。
    请帮我制定一个详细的旅行计划。
    """

    # 执行 LLM + Skills + MCP 协同规划
    report = await demo.run(user_request)

    print("\n" + "=" * 60)
    print("📋 旅行规划报告")
    print("=" * 60)
    print(report)


if __name__ == "__main__":
    # 检查 API 密钥
    # if not os.getenv("OPENAI_API_KEY"):
    #     print("❌ 请设置 OPENAI_API_KEY 环境变量")
    #     print("示例: export OPENAI_API_KEY='sk-...'")
    #     exit(1)

    asyncio.run(main())

模型调用

import os
from openai import OpenAI
from typing import List, Dict, Any
import json


class LLMClient:
    """LLM 客户端（使用 GPT 或兼容 API）"""
    # deepseek-ai/DeepSeek-R1-Distill-Qwen-7B
    def __init__(self, api_key=None, base_url=None, model="deepseek-ai/DeepSeek-V3.2"):
        self.client = OpenAI(
            api_key=api_key or os.getenv("OPENAI_API_KEY", "sk-xxxxxxxxx"),
            base_url=base_url or os.getenv("OPENAI_BASE_URL", "https://api.siliconflow.cn/v1")
        )
        self.model = model

    async def plan_travel(self, user_request: str, available_tools: List[Dict]) -> str:
        """使用 LLM 规划旅行（核心决策逻辑）"""

        # 构建系统提示词（包含技能和可用工具信息）
        system_prompt = self._build_system_prompt(available_tools)

        # 调用 LLM 生成旅行计划
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_request}
            ],
            temperature=0.3
        )

        return response.choices[0].message.content

    def _build_system_prompt(self, tools: List[Dict]) -> str:
        """构建包含技能和工具信息的系统提示词"""

        # 加载技能说明
        skill_path = "skills/travel_plan/SKILL.md"
        with open(skill_path, 'r', encoding='utf-8') as f:
            skill_instruction = f.read()

        # 工具描述
        tools_desc = "\n".join([f"- {tool['name']}: {tool['description']}" for tool in tools])

        return f"""
你是一位专业的旅行规划师，请根据用户需求制定个性化旅行计划。

## 技能说明
{skill_instruction}

## 可用工具
{tools_desc}

## 规划要求
1. 分析用户需求，确定关键参数（目的地、时间、预算、兴趣）
2. 智能选择并调用相关工具获取实时信息
3. 整合航班、酒店、天气、景点信息生成完整计划
4. 按天安排行程，考虑时间合理性和用户体验
5. 提供预算估算和实用建议

## 输出格式
请直接生成完整的旅行计划报告，包含：
- 行程概览
- 每日详细安排
- 预算估算
- 实用建议
"""

skill配置

# 智能旅行规划技能
你是一位专业的旅行规划师，擅长根据用户需求制定个性化旅行计划。

## 核心能力
- 分析用户旅行需求（目的地、时间、预算、兴趣）
- 智能调用相关工具获取实时信息
- 整合多源数据生成最优旅行方案

## 规划流程
1. 分析用户需求 → 确定关键参数
2. 调用航班工具 → 搜索最佳航班
3. 调用酒店工具 → 筛选合适住宿
4. 调用天气工具 → 考虑天气因素
5. 调用景点工具 → 安排每日行程
6. 整合信息 → 生成完整旅行计划

## 输出要求
- 包含航班、酒店、天气、景点等完整信息
- 按天安排行程，考虑地理位置和时间合理性
- 提供预算估算和实用建议
- 对特殊需求（如亲子、老人）给予特别考虑
- 严格使用mcp工具中的数据，不准捏造或者参考历史数据

MCP工具

from mcp.server import Server
from mcp.types import TextContent, Tool
import httpx
import json

server = Server("travel-tools")
from mcp.server import Server
from mcp.types import TextContent, Tool


def create_server():
    """创建 MCP 服务器的工厂函数"""
    server = Server("travel-tools")

    @server.list_tools()
    async def list_tools():
        return [
            Tool(
                name="search_flights",
                description="搜索航班信息，参数：departure_city(出发城市), arrival_city(到达城市), date(日期 YYYY-MM-DD)"
            ),
            Tool(
                name="search_hotels",
                description="搜索酒店信息，参数：city(城市), check_in_date(入住日期), check_out_date(离店日期)"
            ),
            # ... 其他工具定义
        ]

    @server.call_tool()
    async def call_tool(tool_name: str, arguments: dict):
        # ... 工具实现代码
        pass

    return server

from pydantic import BaseModel
# 创建服务器实例
server = create_server()
# 定义工具参数模型
class SearchFlightsParams(BaseModel):
    departure_city: str
    arrival_city: str
    date: str

class SearchHotelsParams(BaseModel):
    city: str
    check_in_date: str
    check_out_date: str

class WeatherParams(BaseModel):
    city: str
    date: str

class AttractionsParams(BaseModel):
    city: str
    category: str
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="search_flights",
            description="搜索航班信息，参数：departure_city(出发城市), arrival_city(到达城市), date(日期 YYYY-MM-DD)",
            inputSchema=SearchFlightsParams.model_json_schema()
        ),
        Tool(
            name="search_hotels",
            description="搜索酒店信息，参数：city(城市), check_in_date(入住日期), check_out_date(离店日期)",
            inputSchema=SearchHotelsParams.model_json_schema()
        ),
        Tool(
            name="get_weather_forecast",
            description="获取天气预报，参数：city(城市), date(日期)",
            inputSchema=WeatherParams.model_json_schema()
        ),
        Tool(
            name="search_attractions",
            description="搜索旅游景点，参数：city(城市), category(类别: 自然/文化/娱乐)",
            inputSchema=AttractionsParams.model_json_schema()
        )
    ]


@server.call_tool()
async def call_tool(tool_name: str, arguments: dict):
    try:
        if tool_name == "search_flights":
            departure = arguments.get("departure_city", "北京")
            arrival = arguments.get("arrival_city", "上海")
            date = arguments.get("date", "2026-02-15")

            # 模拟航班数据
            flights = [
                {"airline": "东方航空", "flight_no": "MU5101", "departure_time": "08:00", "arrival_time": "10:30",
                 "price": 680},
                {"airline": "中国国航", "flight_no": "CA1501", "departure_time": "14:20", "arrival_time": "16:45",
                 "price": 720}
            ]
            flights_text = "\n".join(
                [f"{f['airline']} {f['flight_no']}: {f['departure_time']}-{f['arrival_time']} ¥{f['price']}" for f in
                 flights])
            return [TextContent(type="text", text=f"{departure}→{arrival} 航班信息({date}):\n{flights_text}")]

        elif tool_name == "search_hotels":
            city = arguments.get("city", "上海")
            check_in = arguments.get("check_in_date", "2026-02-15")
            check_out = arguments.get("check_out_date", "2026-02-17")

            # 模拟酒店数据
            hotels = [
                {"name": "上海外滩茂悦大酒店", "price": 850, "rating": 4.7},
                {"name": "上海静安洲际酒店", "price": 680, "rating": 4.5},
                {"name": "上海豫园万丽酒店", "price": 520, "rating": 4.3}
            ]
            hotels_text = "\n".join([f"{h['name']} - 评分{h['rating']} - ¥{h['price']}/晚" for h in hotels])
            return [TextContent(type="text", text=f"{city}酒店信息({check_in}至{check_out}):\n{hotels_text}")]

        elif tool_name == "get_weather_forecast":
            city = arguments.get("city", "上海")
            date = arguments.get("date", "2026-02-15")

            # 模拟天气数据
            weather_map = {
                "2026-02-15": {"temp": "18°C/12°C", "condition": "多云"},
                "2026-02-16": {"temp": "20°C/14°C", "condition": "晴朗"},
                "2026-02-17": {"temp": "19°C/13°C", "condition": "小雨"}
            }
            weather = weather_map.get(date, {"temp": "20°C/15°C", "condition": "晴朗"})
            return [TextContent(type="text", text=f"{city}天气({date}): {weather['temp']} {weather['condition']}")]

        elif tool_name == "search_attractions":
            city = arguments.get("city", "上海")
            category = arguments.get("category", "文化")

            # 模拟景点数据
            attractions = {
                "文化": ["外滩", "豫园", "上海博物馆", "田子坊"],
                "自然": ["世纪公园", "滨江森林公园", "佘山国家森林公园"],
                "娱乐": ["迪士尼乐园", "海昌海洋公园", "欢乐谷"]
            }
            attrs = attractions.get(category, ["外滩", "豫园", "上海博物馆"])
            return [TextContent(type="text", text=f"{city}{category}类景点: {', '.join(attrs)}")]

        else:
            raise ValueError(f"未知工具: {tool_name}")

    except Exception as e:
        return [TextContent(type="text", text=f"工具调用失败: {str(e)}")]

# from mcp.server.stdio import stdio_server
# import asyncio
# # 3. 启动服务
# async def main():
#     async with stdio_server() as (read_stream, write_stream):
#         await server.run(
#             read_stream,
#             write_stream,
#             server.create_initialization_options()
#         )
#
# if __name__ == "__main__":
#     asyncio.run(main())

运行结果

🚀 开始智能旅行规划...
📝 用户需求: 
    我想规划一个3天2晚的上海之旅，时间定在2月15日-17日。
    预算大概3000元左右，希望住在市中心交通方便的地方。
    我对文化历史景点比较感兴趣，还想体验当地美食。
    请帮我制定一个详细的旅行计划。
    
🛠️ 可用工具: ['search_flights', 'search_hotels', 'get_weather_forecast', 'search_attractions']
🧠 LLM 正在分析需求并制定计划...
🔧 执行计划并获取实时信息...

============================================================
📋 旅行规划报告
============================================================

🌟 智能旅行规划完成！

以下是基于你的需求生成的个性化旅行计划：

好的，收到您的需求！作为一名专业的旅行规划师，我将根据您提供的信息——**上海、3天2晚（2月15-17日）、预算约3000元、偏好文化历史与美食、住宿需市中心交通便利**——为您制定一份个性化的旅行计划。

首先，我将调用相关工具，为您获取实时的航班、酒店、天气和景点信息。

### 第一步：信息收集与分析

1.  **获取天气预报**：了解旅行期间的天气状况，以便合理安排行程和着装。
2.  **搜索酒店**：根据您的入住日期（2月15-16日两晚）和市中心交通便利的要求，筛选合适酒店。
3.  **搜索景点**：根据您对文化历史的兴趣，搜索相关景点。
4.  **（假设）航班信息**：由于您未指定出发城市，我无法为您查询具体航班。本计划将默认您于2月15日上午抵达上海，2月17日晚间或下午离开。**在实际规划时，请务必使用您的出发城市查询具体航班。**

现在，开始调用工具获取数据。

```json
{
  "city": "上海",
  "date": "2025-02-15"
}
```

```json
{
  "city": "上海",
  "check_in_date": "2025-02-15",
  "check_out_date": "2025-02-17"
}
```

```json
{
  "city": "上海",
  "category": "文化"
}
```

基于以上获取的实时信息，我为您整合并制定以下旅行计划。

---

### **上海文化美食三日游个性化计划 (2025年2月15日-17日)**

#### **一、 行程概览**
*   **主题**：海派文化寻踪与舌尖体验
*   **天数**：3天2晚
*   **核心区域**：黄浦区、徐汇区
*   **天气提示**：根据预报，2月15日多云，16-17日有小雨。气温在4-9°C之间，体感湿冷。**请务必携带保暖衣物、围巾手套，并准备好雨具。**
*   **住宿建议**：基于搜索，建议选择**人民广场、南京东路、静安寺或陕西南路**地铁站附近的酒店。这些区域是轨道交通枢纽，前往各大景点极为方便，周边餐饮购物选择也极多。经济型或舒适型酒店价格约在400-600元/晚。

#### **二、 每日详细行程安排**

**第一天 (2月15日，周六) ： 城市脉络·外滩万国建筑博览**
*   **上午**：抵达上海。乘坐地铁2号线或机场大巴前往市中心酒店办理入住（假设住人民广场附近）。
*   **午餐**：在酒店附近的“老字号”解决，如**大壶春**（生煎）、**德兴馆**（焖蹄面），体验本帮面点。
*   **下午**：**上海博物馆**（免费，需预约）。这里是了解上海及中国古代历史的绝佳窗口，馆藏文物丰富，预计游览3小时。
*   **傍晚**：从博物馆步行至**人民广场**，感受城市绿肺。随后沿**南京路步行街**向东漫步，感受“中华商业第一街”的繁华，直达外滩。
*   **晚上**：**外滩**夜景是上海的名片。华灯初上时，欣赏对岸陆家嘴的摩天楼群与身后百年历史的“万国建筑博览群”形成的震撼对比。晚餐可在南京东路或江西中路一带寻找本帮菜馆，如**上海老饭店**（人均稍高，可点招牌菜）或**老正兴菜馆**。
*   **住宿**：市中心酒店。

**第二天 (2月16日，周日，小雨) ： 法租界漫步·文艺与小资**
*   **上午**：乘坐地铁至**徐家汇**站，参观**徐家汇天主教堂**和**徐家汇书院**，感受上海近代中西文化交汇的起点。
*   **午餐**：在徐家汇商圈解决，选择多样。
*   **下午**：地铁至**常熟路**或**陕西南路**站，开始**衡复历史文化风貌区**（原法租界）漫步。推荐路线：**衡山路**—**东平路**—**桃江路**—**汾阳路**。这里遍布花园洋房、梧桐树，可参观**上海工艺美术博物馆**（“小白宫”），或在路边特色咖啡馆歇脚。
*   **晚上**：前往**新天地**，这里是石库门建筑改造的时尚地标。晚餐可在新天地内或附近的**进贤路**、**茂名南路**，品尝精致的本帮菜或创意菜。饭后可欣赏石库门夜景。
*   **住宿**：市中心酒店。

**第三天 (2月17日，周一，小雨) ： 老城厢记忆·告别之旅**
*   **上午**：前往**豫园**及**城隍庙**商业区。豫园是经典的江南园林，城隍庙则充满了老上海民俗风情。可以品尝**南翔馒头店**的小笼包（通常排队较长）、**宁波汤团店**的汤圆等特色小吃。
*   **午餐**：在城隍庙美食广场解决，一站式品尝多种上海风味小吃。
*   **下午**：根据航班或列车时间，可选择在酒店附近最后采购一些特产如**国际饭店的蝴蝶酥**、**沈大成的糕团**，然后前往机场/火车站，结束愉快的旅程。

#### **三、 预算估算 (每人，人民币)**
*   **住宿**：舒适型酒店 500元/晚 x 2晚 = **1，000元**
*   **餐饮**：每日约150元（早餐30+午餐60+晚餐60）x 3天 = **450元** (包含小吃、咖啡等)
*   **市内交通**：以地铁为主，每日约15元 x 3天 = **45元**
*   **景点门票**：豫园旺季门票40元，上海工艺美术博物馆8元。其他主要景点（博物馆、外滩、风貌区）免费。总计约 **50元**
*   **购物/应急**：预留 **300元**
*   **小计**：1000 + 450 + 45 + 50 + 300 = **1，845元**
*   **大交通（往返机票/火车票）**：**此项浮动最大，需您自行查询确认。** 从国内主要城市往返上海，预算需预留 **800 - 1200元**。
*   **总预算范围**：**2，645元 - 3，045元** （控制在您3000元预算内）

#### **四、 实用建议**
1.  **交通**：提前在手机上下载“Metro大都会”APP，扫码乘坐地铁，非常方便。市内主要靠地铁+步行即可。
2.  **预约**：**上海博物馆**、**徐家汇书院**等热门场馆务必提前在其官方微信公众号上预约。
3.  **美食**：除了计划内的，可以随时关注路边人气小店。早餐尝试“四大金刚”（大饼、油条、粢饭、豆浆），夜宵可以尝尝“耳光馄饨”。
4.  **着装**：2月上海湿冷，室内外温差大，建议采用“洋葱式”穿法，便于穿脱。一双舒适防水的步行鞋至关重要。
5.  **支付**：移动支付（微信/支付宝）已全面普及，无需携带大量现金。

**最后提醒**：本计划基于当前可获取的实时信息生成。在出发前，请再次通过工具确认**航班、酒店**的具体信息，并关注**天气**的最终预报。祝您在上海有一段难忘的文化美食之旅！

---
📊 信息说明：
- 航班、酒店、天气信息均为实时模拟数据
- 行程安排已考虑时间合理性和地理位置
- 预算估算包含交通、住宿、门票等主要费用