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())