LLM+MCP工具调用
MCP “模型上下文协议”（Model Context Protocol），是一套用于规范 “大模型（LLM）与外部工具” 交互的标准化机制。
本文通过C/S模式实现LLM（阿里百炼）+MCP工具调用，如高德天气预报；结尾Java为应用示例
Client
# blade_mcp_api.py
import os
import json
import requests
from dotenv import load_dotenv
from dashscope import Generation
from flask import Flask, request, jsonify# 加载环境变量
load_dotenv()
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL")# 验证配置是否完整
if not DASHSCOPE_API_KEY:raise ValueError("请在.env文件中配置DASHSCOPE_API_KEY")
if not MCP_SERVER_URL:raise ValueError("请在.env文件中配置MCP_SERVER_URL")app = Flask(__name__)class BladeMCP:def __init__(self):self.tools_desc = """可用工具说明：1. 工具名：get_weather，功能：获取指定城市天气，参数：city（字符串，如"深圳"）2. 工具名：translate，功能：文本翻译，参数：text（待翻译文本）、target_lang（目标语言，如"en"代表英文、"zh"代表中文）调用要求：- 若需要调用工具，仅返回JSON格式（无其他多余文字），示例：{"tool_name": "get_weather", "arguments": {"city": "深圳"}}- 若不需要调用工具，直接用自然语言回答用户问题"""def call_mcp_server(self, tool_name: str, **kwargs) -> str:"""调用MCP服务器的工具接口，返回执行结果"""try:response = requests.post(url=f"{MCP_SERVER_URL}/mcp/call",json={"tool_name": tool_name, "arguments": kwargs},timeout=15,headers={"Content-Type": "application/json"})response.raise_for_status()result_data = response.json()return result_data["result"] if result_data["status"] == "success" else f"MCP工具错误：{result_data.get('detail', '未知错误')}"except requests.exceptions.Timeout:return "MCP调用超时：服务器未在15秒内响应"except requests.exceptions.ConnectionError:return f"MCP连接失败：无法连接到{MCP_SERVER_URL}，请检查服务器是否已启动"except Exception as e:return f"MCP调用异常：{str(e)}"def process_user_query(self, user_query: str) -> dict:"""处理用户查询，返回结构化的结果"""prompt = f"""用户当前问题：{user_query}请根据以下工具说明处理：{self.tools_desc}注意：严格按照要求返回，不要添加额外解释文字"""try:llm_response = Generation.call(model="qwen-turbo",messages=[{"role": "user", "content": prompt}],api_key=DASHSCOPE_API_KEY,temperature=0.3)llm_reply = llm_response.output["text"].strip()print(f"[LLM原始回复]：{llm_reply}")if "{" in llm_reply and "}" in llm_reply:try:tool_call = json.loads(llm_reply)tool_name = tool_call.get("tool_name")tool_args = tool_call.get("arguments", {})if not tool_name:return {"status": "error", "message": "LLM工具调用错误：未指定tool_name"}if tool_name == "get_weather" and "city" not in tool_args:return {"status": "error", "message": "LLM工具调用错误：get_weather需传入city参数"}if tool_name == "translate" and not ("text" in tool_args and "target_lang" in tool_args):return {"status": "error", "message": "LLM工具调用错误：translate需传入text和target_lang参数"}print(f"[调用MCP工具]：{tool_name}，参数：{tool_args}")mcp_result = self.call_mcp_server(tool_name, **tool_args)print(f"[MCP工具结果]：{mcp_result}")final_prompt = f"""用户问题：{user_query}工具调用结果：{mcp_result}请基于工具结果，用简洁自然的语言回答用户，不要添加额外内容"""final_llm_response = Generation.call(model="qwen-turbo",messages=[{"role": "user", "content": final_prompt}],api_key=DASHSCOPE_API_KEY,temperature=0.5)return {"status": "success","type": "tool_call","tool_used": tool_name,"tool_result": mcp_result,"final_answer": final_llm_response.output['text'].strip()}except json.JSONDecodeError:return {"status": "error", "message": f"LLM回复格式错误：无法解析JSON，原始回复：{llm_reply}"}except Exception as e:return {"status": "error", "message": f"工具调用处理异常：{str(e)}"}else:return {"status": "success","type": "direct_answer","final_answer": llm_reply}except Exception as e:return {"status": "error", "message": f"LLM调用异常：{str(e)}"}# 初始化实例
blade_mcp = BladeMCP()@app.route('/api/query', methods=['POST'])
def handle_query():"""处理查询的API接口"""try:data = request.get_json()if not data or 'query' not in data:return jsonify({"status": "error", "message": "缺少query参数"}), 400user_query = data['query']print(f"[API请求] 用户查询: {user_query}")result = blade_mcp.process_user_query(user_query)print(f"[API响应] 处理结果: {result}")return jsonify(result)except Exception as e:error_msg = f"服务器内部错误：{str(e)}"print(f"[API错误] {error_msg}")return jsonify({"status": "error", "message": error_msg}), 500@app.route('/health', methods=['GET'])
def health_check():"""健康检查接口"""return jsonify({"status": "healthy","service": "BladeMCP API","mcp_server": MCP_SERVER_URL})@app.route('/')
def root():"""根路径"""return jsonify({"message": "BladeMCP API服务运行中","endpoints": {"健康检查": "/health","查询接口": "/api/query (POST)"}})def start_server(host="0.0.0.0", port=5000):"""启动服务器函数"""print("=" * 50)print("阿里百炼LLM + MCP服务 API服务器")print(f"当前使用的API Key：{DASHSCOPE_API_KEY[:10]}...")print(f"当前连接的MCP服务器：{MCP_SERVER_URL}")print(f"服务地址: http://{host}:{port}")print("=" * 50 + "\n")# 启动Flask服务器app.run(host=host, port=port, debug=False)if __name__ == "__main__":# 直接启动服务器start_server()Server
# mcp_http_server.py
import os
import requests
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModelapp = FastAPI(title="MCP HTTP服务")# 加载环境变量
load_dotenv()
AMAP_WEATHER_KEY = os.getenv("AMAP_WEATHER_KEY")
AMAP_WEATHER_URL = os.getenv("AMAP_WEATHER_URL")# 验证环境变量
if not AMAP_WEATHER_KEY:print("警告: 未设置AMAP_WEATHER_KEY环境变量，天气功能可能无法正常工作")
if not AMAP_WEATHER_URL:AMAP_WEATHER_URL = "https://restapi.amap.com/v3/weather/weatherInfo"print(f"使用默认天气API URL: {AMAP_WEATHER_URL}")# 定义工具调用请求格式
class ToolCallRequest(BaseModel):tool_name: strarguments: dict = {}# 注册MCP工具
class MCPTools:@staticmethoddef get_weather(city: str) -> str:"""调用高德天气API"""try:if not AMAP_WEATHER_KEY:return "天气服务暂不可用：未配置API密钥"# 1.调用高德API获取城市编码adcode = MCPTools._get_city_adcode(city)if not adcode:return f"无法找到城市'{city}'的编码，请检查城市名称是否正确"# 2.通过adcode查询天气params = {"key": AMAP_WEATHER_KEY,"city": adcode,"extensions": "base"}response = requests.get(AMAP_WEATHER_URL, params=params, timeout=10)response.raise_for_status()weather_data = response.json()# 3.解析结果if weather_data["status"] != "1":return f"天气查询失败：{weather_data.get('info', '未知错误')}"lives = weather_data.get("lives", [])if not lives:return "无法获取天气信息"weather_info = lives[0]return (f"{city} 实时天气：{weather_info['weather']}，"f"气温{weather_info['temperature']}℃，"f"{weather_info['winddirection']}风{weather_info['windpower']}级，"f"湿度{weather_info['humidity']}%")except requests.exceptions.Timeout:return "天气查询超时，请稍后重试"except requests.exceptions.ConnectionError:return "天气查询失败：请检查网络连接"except Exception as e:return f"天气查询异常：{str(e)}"@staticmethoddef _get_city_adcode(city: str) -> str:"""辅助方法：通过城市名查询高德adcode（行政编码）"""try:adcode_url = "https://restapi.amap.com/v3/config/district"params = {"key": AMAP_WEATHER_KEY,"keywords": city,"subdistrict": 0 # 不返回子区域}response = requests.get(adcode_url, params=params, timeout=10)response.raise_for_status()district_data = response.json()if district_data["status"] == "1" and district_data["districts"]:return district_data["districts"][0]["adcode"] # 返回第一个匹配的adcodereturn ""except Exception as e:print(f"获取城市编码失败: {e}")return ""@staticmethoddef translate(text: str, target_lang: str) -> str:"""翻译文本到目标语言（模拟）"""# 扩展翻译词典translations = {"hello": {"zh": "你好", "fr": "bonjour", "ja": "こんにちは"},"世界": {"en": "world", "fr": "monde", "ja": "世界"},"我爱编程": {"en": "I love programming", "fr": "J'aime programmer", "ja": "プログラミングが大好き"},"更爱探索AI工具": {"en": "and love exploring AI tools even more","fr": "et j'aime encore plus explorer les outils d'IA","ja": "AIツールの探索がもっと好き"}}# 尝试直接匹配if text in translations and target_lang in translations[text]:return translations[text][target_lang]# 尝试部分匹配（用于长句子）for key, trans in translations.items():if key in text and target_lang in trans:return trans[target_lang]return f"暂不支持翻译：'{text}' 到 '{target_lang}'，当前支持的语言：中文(zh)、英文(en)、法文(fr)、日文(ja)"# 暴露HTTP接口
@app.post("/mcp/call")
def call_tool(request: ToolCallRequest):try:tool = getattr(MCPTools, request.tool_name, None)if not tool:raise HTTPException(status_code=404, detail=f"工具不存在: {request.tool_name}")print(f"[MCP服务] 调用工具: {request.tool_name}, 参数: {request.arguments}")result = tool(**request.arguments)print(f"[MCP服务] 工具结果: {result}")return {"status": "success", "result": result}except Exception as e:error_msg = f"工具调用失败: {str(e)}"print(f"[MCP服务] {error_msg}")raise HTTPException(status_code=500, detail=error_msg)@app.get("/health")
def health_check():"""健康检查接口"""return {"status": "healthy","service": "MCP HTTP服务","available_tools": ["get_weather", "translate"]}@app.get("/")
def root():"""根路径"""return {"message": "MCP HTTP服务运行中","version": "1.0.0","endpoints": {"健康检查": "/health","工具调用": "/mcp/call (POST)"}}def start_server(host="0.0.0.0", port=8000):"""启动服务器函数"""print("=" * 50)print("MCP HTTP服务启动中...")print(f"服务地址: http://{host}:{port}")print(f"健康检查: http://{host}:{port}/health")print("可用工具: get_weather, translate")print("=" * 50)# 启动uvicorn服务器uvicorn.run(app,host=host,port=port,log_level="info")if __name__ == "__main__":# 直接启动服务器start_server()Java应用示例
import com.fasterxml.jackson.databind.ObjectMapper;
import okhttp3.*;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;public class BladeMCPClient {private static final String API_BASE_URL = "http://localhost:5000";private static final OkHttpClient client = new OkHttpClient();private static final ObjectMapper mapper = new ObjectMapper();private static final MediaType JSON = MediaType.parse("application/json; charset=utf-8");/*** 发送查询到BladeMCP服务*/public static Map<String, Object> sendQuery(String userQuery) throws IOException {// 构建请求体Map<String, String> requestBody = new HashMap<>();requestBody.put("query", userQuery);String jsonBody = mapper.writeValueAsString(requestBody);Request request = new Request.Builder().url(API_BASE_URL + "/api/query").post(RequestBody.create(jsonBody, JSON)).build();try (Response response = client.newCall(request).execute()) {if (!response.isSuccessful()) {throw new IOException("HTTP错误: " + response.code());}String responseBody = response.body().string();return mapper.readValue(responseBody, Map.class);}}/*** 健康检查*/public static boolean healthCheck() throws IOException {Request request = new Request.Builder().url(API_BASE_URL + "/health").get().build();try (Response response = client.newCall(request).execute()) {return response.isSuccessful();}}public static void main(String[] args) {try {System.out.println("=== BladeMCP Java客户端测试 ===");// 健康检查boolean isHealthy = healthCheck();System.out.println("服务健康状态: " + (isHealthy ? "正常" : "异常"));if (!isHealthy) {System.out.println("请确保Python服务正在运行在端口8000");return;}// 测试查询String[] testQueries = {"吉林市今天的天气怎么样？","hello,world","先有鸡还是先有蛋"};for (int i = 0; i < testQueries.length; i++) {System.out.println("\n=== 测试 " + (i + 1) + " ===");System.out.println("用户查询: " + testQueries[i]);Map<String, Object> result = sendQuery(testQueries[i]);// 打印结果System.out.println("状态: " + result.get("status"));System.out.println("类型: " + result.get("type"));if ("success".equals(result.get("status"))) {if ("tool_call".equals(result.get("type"))) {System.out.println("使用的工具: " + result.get("tool_used"));System.out.println("工具结果: " + result.get("tool_result"));}System.out.println("最终回答: " + result.get("final_answer"));} else {System.out.println("错误信息: " + result.get("message"));}}} catch (Exception e) {System.err.println("客户端错误: " + e.getMessage());e.printStackTrace();}}
}