外部调用

pip install openai mcp

# -*- coding: utf-8 -*-
# 使用 OpenAI SDK + MCP SDK 调用千问AI平台联网搜索增强版（EnhancedSearch）MCP 服务
import os
import asyncio
import json
from openai import OpenAI
from mcp.client.streamable_http import streamablehttp_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://dashscope.aliyuncs.com/api/v1/mcps/EnhancedSearch/mcp"
    headers = {"Authorization": f"Bearer {api_key}"}
    # 1. 连接 MCP Server，获取可用工具列表
    async with streamablehttp_client(mcp_url, headers=headers) as (read, write, _):
        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.inputSchema or {"type": "object", "properties": {}},
                    },
                })
            # 2. 调用 DashScope（OpenAI 兼容接口）
            client = OpenAI(
                api_key=api_key,
                base_url="https://dashscope.aliyuncs.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,
                    })

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 灵活编码集成至自有项目。