AI知识科普(8):在实战中掌握MCP——AI的触手,它并不神秘!

import httpxdef get_tool_list():    url = "https://mcp-weathertrax.jaredco.com/mcp"    payload = {        "jsonrpc": "2.0",        "method": "tools/list",        "id": 1    }    with httpx.Client() as client:        resp = client.post(url, json=payload)        data = resp.json()    if "error" in data:        print(f"调用失败: {data['error']}")        return    tools = data.get("result", {}).get("tools", [])    for tool in tools:        print(f"工具名称: {tool.get('name')}")        print(f"工具描述: {tool.get('description', '')}")        print(f"工具参数: {tool.get('inputSchema', {})}")        print()if __name__ == "__main__":    get_tool_list()

payload = {    "jsonrpc": "2.0",    "method": "tools/call",    "params": {        "name": "weatherTool",        "arguments": {            "location": "北京",            "query_type": "current"        }    },    "id": 2}

import jsonimport httpxclass MCPClient:    """ 通用MCP-CLIENT类 - HTTPx版本 """    def __init__(self, s_url: str):        self.s_url = s_url        self.tools = []        self._client = None    def connect(self):        """ 建立连接并获取工具列表 """        # 建立连接        self._client = httpx.Client()        # 获取工具列表        payload = {            "jsonrpc": "2.0",            "method": "tools/list",            "id": 1        }        resp = self._client.post(self.s_url, json=payload)        data = resp.json()        if "result" not in data or "tools" not in data["result"]:            raise RuntimeError("get tool list fail!")        # 记录工具列表        for tool in data["result"]["tools"]:            self.tools.append({                "type": "function",                "function": {                    "name": tool["name"],                    "description": tool.get("description", ""),                    "parameters": tool.get("inputSchema", {})                }            })        return True    def call(self, name: str, arguments: dict):        """ 调用工具 """        if self._client is None:            raise RuntimeError("no MCP connect!")        # 调用工具        payload = {            "jsonrpc": "2.0",            "method": "tools/call",            "params": {"name": name, "arguments": arguments},            "id": 1        }        resp = self._client.post(self.s_url, json=payload)        data = resp.json()        # 解析结果        if "error" in data:            return f"错误: {data['error']}"        if "result" in data:            result = data["result"]            # 尝试提取文本内容            if isinstance(result, dict) and "content" in result:                texts = []                for item in result["content"]:                    if isinstance(item, dict) and "text" in item:                        texts.append(item["text"])                if texts:                    return '\n'.join(texts)            return json.dumps(result, ensure_ascii=False)        return json.dumps(data, ensure_ascii=False)    def close(self):        """ 关闭连接 """        if self._client:            self._client.close()            self._client = None

# 初始化MCP客户端，并建立连接mcp_client = MCPClient(MCP_SERVER)mcp_client.connect()# 显示MCP工具列表print(f"🤖 MCP Server: \n   {MCP_SERVER}\n")print(f"🤖 MCP Tool-list：")for tool in mcp_client.tools:    print(f"   {tool['function']['name']}")print("\n")# 初始化客户端llm_client = ZhipuAI(api_key="这里填你的API KEY")print("🤖 对话开始（输入 'quit' 或 'exit' 退出）")print("-" * 50)dialog = []while True:    # 获取用户输入    user_input = input("\n👤 你: ").strip()    # 判断退出条件    if user_input.lower() in ['quit', 'exit']:        print("\n👋 再见！\n")        break;    # 跳过空输入    if not user_input:        continue    # 将新对话内容追加到dialog尾部    dialog.append({"role": "user", "content": user_input})    try:        start_time = time.time()        # 调用大模型        response = llm_client.chat.completions.create(            model="glm-4-flash",            messages=dialog,            tools=mcp_client.tools,            tool_choice="auto",            stream=True        )        # 计算响应时延        end_time = time.time()        print(f"\n⚡ 响应延迟:【{(end_time - start_time)*1000:.0f}ms】\n")        print("🤖 AI: ", end="", flush=True)        full_reply = ""        tool_lists = []        for chunk in response:            delta = chunk.choices[0].delta            # 收集完整回复            if delta.content:                print(delta.content, end="", flush=True)                full_reply += delta.content            # 收集tool_calls            if delta.tool_calls:                tool_lists = delta.tool_calls        # 调用MCP工具        if tool_lists:            tool_calls = []            for tool in tool_lists:                tool_calls.append({                    "id": tool.id,                    "type": "function",                    "function": {                        "name": tool.function.name,                        "arguments": tool.function.arguments                    }                })            tool_call = tool_lists[0]            tool_name = tool_call.function.name            tool_args = json.loads(tool_call.function.arguments)            tool_rets = mcp_client.call(tool_name, tool_args)            print(f"🔧 [调用 MCP 工具] {tool_name}({tool_args})")            print("🤖 AI: ", end="", flush=True)            # 调用MCP工具后，发起第二次大模型调用，让大模型做出最终答复            response = llm_client.chat.completions.create(                model="glm-4-flash",                messages=[                    {"role": "user", "content": user_input},                    {"role": "assistant", "content": full_reply, "tool_calls": tool_calls},                    {"role": "tool", "tool_call_id": tool_call.id, "content": tool_rets}                ],                stream=True            )            full_reply = ""            for chunk in response:                if chunk.choices and chunk.choices[0].delta.content:                    content = chunk.choices[0].delta.content                    print(content, end="", flush=True)                    full_reply += content            print("\n")        # 把AI回复的内容添加到会话列表里        dialog.append({"role": "assistant", "content": full_reply})    except Exception as e:        print(f"❌ 错误: {e}")# 关闭MCP连接mcp_client.close()