AI-agent/04_mcp_homework/step5_qwen_agent_use_mcp.py

"""
第 5 步：让 Qwen Agent 自己决定是否调用 MCP 工具。

前面 step2/step3 是我们在代码里手动调用工具：
    tool_map["search_hotels"].ainvoke(...)

这一步变成：
    用户用自然语言提问
    Qwen 判断该调用 search_transport 还是 search_hotels
    MCP Server 执行工具
    Qwen 根据工具结果回答

运行：
python step5_qwen_agent_use_mcp.py
"""

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI


def build_qwen_model() -> ChatOpenAI:
    load_dotenv()

    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise RuntimeError("请先在 .env 里填写 DASHSCOPE_API_KEY。")

    return ChatOpenAI(
        api_key=api_key,
        base_url=os.getenv("QWEN_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
        model=os.getenv("QWEN_MODEL", "qwen-plus"),
        temperature=0.2,
    )


async def main() -> None:
    server_file = Path(__file__).with_name("my_mcp_demo.py").resolve()

    client = MultiServerMCPClient(
        {
            "travel-tools": {
                "transport": "stdio",
                "command": sys.executable,
                "args": [str(server_file)],
            }
        }
    )

    tools = await client.get_tools()
    print("Qwen Agent 可用的 MCP 工具：", [tool.name for tool in tools])

    model = build_qwen_model()

    agent = create_agent(
        model=model,
        tools=tools,
        system_prompt=(
            "你是一个旅行助手。"
            "当用户询问住宿时，优先使用 search_hotels 工具；"
            "当用户询问城市间交通时，优先使用 search_transport 工具；"
            "拿到工具结果后，用中文自然语言回答。"
        ),
    )

    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "我想 2026-08-15 从成都去重庆玩 3 天 2 晚，帮我看看交通和住宿怎么安排。",
                }
            ]
        }
    )

    print("\nQwen Agent 最终回答：")
    print(result["messages"][-1].content)


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