Python调用DeepSeek API查询ClickHouse的流程步骤

pip install mcp openai

pip install uv

import asyncio
import os
import json
import sys
from typing import List, Dict, Any, Optional

# 导入 MCP 相关库
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import mcp.types as types

# 导入 OpenAI SDK (DeepSeek 兼容 OpenAI 格式)
from openai import AsyncOpenAI

# ================= 配置部分 =================

# DeepSeek API 配置
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "<YOUR_DEEPSEEK_API_KEY>")
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
MODEL_NAME = "deepseek-chat"  # 或者 "deepseek-reasoner" (R1)

# MCP ClickHouse Server 配置 (来自你的 JSON)
MCP_CLICKHOUSE_CONFIG = {
    "command": "uv",
    "args": [
        "run",
        "--with",
        "mcp-clickhouse",
        "--python",
        "3.10",
        "mcp-clickhouse"
    ],
    "env": {
        "CLICKHOUSE_HOST": "<clickhouse-host>",
        "CLICKHOUSE_PORT": "<clickhouse-port>",
        "CLICKHOUSE_USER": "<clickhouse-user>",
        "CLICKHOUSE_PASSWORD": "<clickhouse-password>",
        "CLICKHOUSE_ROLE": "<clickhouse-role>",
        "CLICKHOUSE_SECURE": "true",
        "CLICKHOUSE_VERIFY": "true",
        "CLICKHOUSE_CONNECT_TIMEOUT": "30",
        "CLICKHOUSE_SEND_RECEIVE_TIMEOUT": "30"
    }
}

# ================= 工具函数 =================

def convert_mcp_tool_to_openai(mcp_tool: types.Tool) -> Dict[str, Any]:
    """
    将 MCP 工具定义转换为 OpenAI/DeepSeek 支持的 Function Calling 格式。
    """
    return {
        "type": "function",
        "function": {
            "name": mcp_tool.name,
            "description": mcp_tool.description,
            "parameters": mcp_tool.inputSchema
        }
    }

# ================= 主逻辑 =================

async def run_chat_loop():
    # 1. 初始化 DeepSeek 客户端
    client = AsyncOpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL)
    
    # 2. 准备 MCP Server 参数
    server_params = StdioServerParameters(
        command=MCP_CLICKHOUSE_CONFIG["command"],
        args=MCP_CLICKHOUSE_CONFIG["args"],
        env={**os.environ, **MCP_CLICKHOUSE_CONFIG["env"]} # 合并当前环境变量
    )

    print(f"正在启动 MCP Server: {MCP_CLICKHOUSE_CONFIG['command']}...")
    
    # 3. 连接 MCP Server 并开始对话循环
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            
            # 获取 MCP Server 提供的工具列表
            mcp_tools_list = await session.list_tools()
            openai_tools = [convert_mcp_tool_to_openai(tool) for tool in mcp_tools_list.tools]
            
            print(f"\n已连接 MCP Server，加载了 {len(openai_tools)} 个工具: {[t['function']['name'] for t in openai_tools]}")
            print("-" * 50)
            print("你可以开始询问关于 ClickHouse 的问题了 (输入 'quit' 退出)。")

            messages = [
                {"role": "system", "content": "你是一个智能助手，可以利用工具查询 ClickHouse 数据库来回答用户问题。"}
            ]

            while True:
                user_input = input("\n用户: ")
                if user_input.lower() in ["quit", "exit"]:
                    break

                messages.append({"role": "user", "content": user_input})

                # 第一轮调用：发送用户问题 + 工具定义给 DeepSeek
                try:
                    response = await client.chat.completions.create(
                        model=MODEL_NAME,
                        messages=messages,
                        tools=openai_tools,
                    )
                except Exception as e:
                    print(f"API 调用错误: {e}")
                    continue

                assistant_msg = response.choices[0].message
                messages.append(assistant_msg)

                # 检查 DeepSeek 是否想调用工具
                if assistant_msg.tool_calls:
                    print(f"\n[思考] 模型决定调用工具...")
                    
                    for tool_call in assistant_msg.tool_calls:
                        tool_name = tool_call.function.name
                        tool_args = json.loads(tool_call.function.arguments)
                        
                        print(f"  -> 调用工具: {tool_name}, 参数: {tool_args}")

                        # 执行 MCP 工具
                        try:
                            mcp_result = await session.call_tool(tool_name, arguments=tool_args)
                            
                            # 获取文本结果 (ClickHouse MCP 通常返回文本或 JSON 文本)
                            tool_output = ""
                            if mcp_result.content:
                                for content in mcp_result.content:
                                    if content.type == "text":
                                        tool_output += content.text
                            
                            print(f"  <- 工具返回结果 (前100字符): {tool_output[:100]}...")

                            # 将工具结果添加回对话历史
                            messages.append({
                                "role": "tool",
                                "tool_call_id": tool_call.id,
                                "content": tool_output
                            })

                        except Exception as e:
                            error_msg = f"工具执行出错: {str(e)}"
                            print(f"  ! {error_msg}")
                            messages.append({
                                "role": "tool",
                                "tool_call_id": tool_call.id,
                                "content": error_msg
                            })

                    # 第二轮调用：将工具结果发送回 DeepSeek 获取最终回答
                    final_response = await client.chat.completions.create(
                        model=MODEL_NAME,
                        messages=messages,
                        # 这里依然传入 tools，以防模型需要继续多步调用
                        tools=openai_tools, 
                    )
                    
                    final_content = final_response.choices[0].message.content
                    print(f"\nDeepSeek: {final_content}")
                    messages.append(final_response.choices[0].message)
                
                else:
                    # 如果不需要调用工具，直接输出结果
                    print(f"\nDeepSeek: {assistant_msg.content}")

if __name__ == "__main__":
    # Windows 下通常需要这个策略
    if sys.platform.startswith('win'):
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
        
    asyncio.run(run_chat_loop())