外部调用

pip install openai "mcp>=2.0"

# -*- coding: utf-8 -*-
# 使用 OpenAI SDK + MCP SDK 调用千问AI平台联网搜索增强版（EnhancedSearch）MCP 服务
import os
import asyncio
import json
import httpx2
from openai import OpenAI
from mcp.client.streamable_http import streamable_http_client
from mcp import ClientSession

async def main():
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        print("错误：请设置环境变量 DASHSCOPE_API_KEY")
        return
    mcp_url = "https://maas.qianwenaiapi.com/api/v1/mcps/EnhancedSearch/mcp"
    # mcp>=2.0 通过传入 httpx2.AsyncClient 携带鉴权头
    http_client = httpx2.AsyncClient(
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=httpx2.Timeout(30.0, read=300.0),
    )
    try:
        # 1. 连接 MCP Server，获取可用工具列表
        async with streamable_http_client(mcp_url, http_client=http_client) as streams:
            read, write = streams[0], streams[1]
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools_result = await session.list_tools()
                # 转换为 OpenAI function calling 格式
                openai_tools = []
                for tool in tools_result.tools:
                    openai_tools.append({
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description or "",
                            "parameters": tool.input_schema or {"type": "object", "properties": {}},
                        },
                    })
                # 2. 调用 DashScope（OpenAI 兼容接口）
                client = OpenAI(
                    api_key=api_key,
                    base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
                )
                messages = [{"role": "user", "content": "搜索一下千问AI平台MCP的最新进展"}]
                print("正在联网搜索...")
                print("=" * 50)
                # 3. 多轮工具调用循环
                while True:
                    response = client.chat.completions.create(
                        model="qwen-max",
                        messages=messages,
                        tools=openai_tools or None,
                    )
                    choice = response.choices[0]
                    msg = choice.message
                    if not msg.tool_calls:
                        print(msg.content)
                        break
                    messages.append(msg)
                    for tc in msg.tool_calls:
                        args = json.loads(tc.function.arguments)
                        result = await session.call_tool(tc.function.name, args)
                        tool_content = ""
                        for block in result.content:
                            if hasattr(block, "text"):
                                tool_content += block.text
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tc.id,
                            "content": tool_content,
                        })
    finally:
        await http_client.aclose()

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

正在联网搜索...
==================================================
千问AI平台 MCP（Model Context Protocol）是一项新推出的服务，允许用户在千问AI平台上统一接入和管理 MCP 服务。最新进展包括：
1. 支持通过 MCP 商店快速接入多种 MCP 服务（如联网抓取、联网搜索增强版等）。
2. 提供 Streamable HTTP 协议，支持外部应用通过标准 HTTP 调用。
3. 支持在 Qoder CLI、Codex、Claude Code、Cursor 等客户端中一键或手动配置。
4. 开发者可通过 MCP SDK 灵活编码集成至自有项目。