mcp学习笔记

MCP（Model Context Protocol）是一种由Anthropic推出的开放协议，旨在统一大型语言模型（LLM）与外部数据源/工具之间的交互。其核心组件包括 MCP Client、MCP Server 和 Function Calling 机制，三者协同工作以扩展LLM的能力，使大模型能够调用外部工具。

MCP协议是提示词工程和Function calling技术发展的产物，将如何调用外部工具这个事进行规范和统一。本文使用Python实现了一个简单的mcp demo程序，可实现利用自然语言来操作数据库，发送手机验证码和邮件。

一、MCP基本概念和原理

首先需要了解MCP中几个重要的概念，可参考这里MCP 简介 - MCP 中文文档
mcp中几个要素的关系如下图所示，用户通过mcp client与大模型进行交互，mcp server提供为mcp client提供外部工具。

使用mcp协议调用外部工具的过程如下：

s1:用户向mcp client提出问题；

s2:mcp client将用户的问题和tools列表一同提交给大模型；

s3:大模型会根据用户的问题，从tools列表中取出最相关最适合的工具，构造好参数，向client返回工具名和参数，也有可能找不到；

s4:mcp client调用根据工具名和参数调用对应的接口，可以是本地的函数，也可以是web api，取决于mcp server如何实现；

s5:mcp client将工具的结果提交给大模型；

s6:大模型对工具返回的结果，来回答用户的原始问题；

s7: mcp client将大模型最终的回答呈现给用户。

二、Function calling

Function Calling 允许LLM根据用户请求生成结构化参数（如JSON格式），触发预定义的外部函数或API调用，从而获取实时数据或执行具体操作。例如用户询问天气时，模型提取参数（城市、日期）并调用天气API。

原理：

开发者注册函数名称、参数类型及描述（类似JSON Schema），要符合json schema规范。
模型解析用户请求，匹配适用的函数。
提取关键信息并格式化为结构化参数（如{"location":"北京"}）
外部系统调用函数后，结果返回模型生成最终回答。

Function calling功能需要对模型进行微调训练，使模型能够理解json schema格式语法，为函数构造正确的json格式参数。

目前支持Function calling功能的模型有

OpenAI GPT系列：从GPT-3.5开始集成此功能，2023年就有了
Anthropic Claude：支持与外部服务深度集成。
国产模型：阿里Qwen、DeepSeek v3等通过微调实现类似能力

三、Function calling示例

下面我们使用python写一个示例程序，来演示如何使用function calling功能，实现利用自然语言来操作数据库和访问第三方接口。这里我们使用的模型是qwen3-32b。

python如何调用通义模型可参考阿里百炼的官方API文档。

下面是阿里官方的例子，可作为参考。

import os
from openai import OpenAIclient = OpenAI(# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",api_key=os.getenv("DASHSCOPE_API_KEY"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 填写DashScope SDK的base_url
)tools = [# 工具1 获取当前时刻的时间{"type": "function","function": {"name": "get_current_time","description": "当你想知道现在的时间时非常有用。","parameters": {}  # 因为获取当前时间无需输入参数，因此parameters为空字典}},  # 工具2 获取指定城市的天气{"type": "function","function": {"name": "get_current_weather","description": "当你想查询指定城市的天气时非常有用。","parameters": {  "type": "object","properties": {# 查询天气时需要提供位置，因此参数设置为location"location": {"type": "string","description": "城市或县区，比如北京市、杭州市、余杭区等。"}},"required": ["location"]}}}
]
messages = [{"role": "user", "content": "杭州天气怎么样"}]
completion = client.chat.completions.create(model="qwen-plus",  # 此处以qwen-plus为例，可按需更换模型名称。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/modelsmessages=messages,tools=tools
)print(completion.model_dump_json())

1.获取阿里百炼的api key

这步比较简单不再展开，现在可领取100万的免费tokens额度。

2.mcp_client

代码如下，功能见注释。

import json
import os
from openai import OpenAI
from mcp_server import *# 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
MODEL = "qwen3-32b"  # 模型名称def main():# 参考https://bailian.console.aliyun.com/?tab=api#/api/?type=model&url=https%3A%2F%2Fhelp.aliyun.com%2Fdocument_detail%2F2712576.html&renderType=iframeclient = OpenAI(# 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx",api_key=os.getenv("Bailian_API_Key"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",)messages=[{"role": "system", "content": "You are a helpful assistant."}]while True:query = input("请输入问题：")if query == "exit":breakelif query == "clear":os.system("cls")continueelif query == "reset":messages = [{"role": "system", "content": "You are a helpful assistant."}]continueelse:messages.append({"role": "user", "content": query})completion = client.chat.completions.create(model=MODEL,messages=messages,extra_body={"enable_thinking": False},tools = json.load(open("tools.json", "r", encoding="utf-8")),# 这里加载工具列表# Qwen3模型通过enable_thinking参数控制思考过程（开源版默认True，商业版默认False）# 使用Qwen3开源版模型时，若未启用流式输出，请将下行取消注释，否则会报错# extra_body={"enable_thinking": False},)# 将大模型的结果保存到文件# 这里使用了model_dump_json()方法，返回的是一个json字符串res  = completion.model_dump_json()json.dump(completion.dict(), open("response1.json", "w", encoding="utf-8"), ensure_ascii=False, indent=4)#若大模型的结果中有tool_calls字段，则表示调用了工具if completion.choices[0].finish_reason == 'tool_calls':# 取出tool_calls字段中的第一个工具调用tool_call = completion.choices[0].message.tool_calls[0]py_code = 'ret = ' +  tool_call.function.name + "(**" + tool_call.function.arguments + ")" # ret = func_name(**args) 的方式传递参数exec(py_code,globals()) # 执行代码,将ret变量映射到当前程序的全局变量空间中# 将function calling的结果加入messages中messages.append(completion.choices[0].message.model_dump())# 将工具调用的结果加入messages中messages.append({'role':'tool','content':  str(ret),'tool_call_id': tool_call.id,})  # 提交给大模型completion = client.chat.completions.create(model=MODEL,messages=messages,extra_body={"enable_thinking": False})# 将大模型的结果保存到文件json.dump(completion.dict(), open("response2.json", "w", encoding="utf-8"), ensure_ascii=False, indent=4)# 输出大模型的最终回答if completion.choices[0].message.content:print(completion.choices[0].message.content)messages.append({"role": "assistant", "content": completion.choices[0].message.content})        main()

2. mcp_server

在mcp_server中我们提供了三类接口，具体见附件

一是访问本地sqlite数据库的接口

execute_query(sql, params=None) 执行SQL查询并返回结果
get_all_table_structures 获取SQLite数据库中所有表的结构信息

二是获取本地公网IP和运营商信息的接口

get_public_ip 获取当前公网IP地址
get_ip_details(ip: str) 获取IP详细信息（含城市、运营商）

三是发送信息的接口

send_verification_code(phone_number: str, code: str) 向指定手机号发送短信验证码
sent_email(to,subject,body) 发送邮件

3. 接口说明

这里的所有的接口的函数和参数描述为json_schema格式，保存为tools.json，这个tools描述要提交给大模型接口。json schema语法可参考JSON Schema reference。


[{"type": "function","function": {"name": "execute_query","description": "执行SQL查询并返回结果","parameters": {"type": "object","properties": {"sql": {"type": "string","description": "要执行的SQL语句"},"params": {"type": "array","description": "SQL语句的参数","items": {"type": "string"}}},"required": ["sql","params"]}}},{"type": "function","function": {"name": "get_all_table_structures","description": "获取所有表的结构信息","parameters": {}}},{"type": "function","function": {"name": "get_public_ip","description": "获取当前公网IP地址","parameters": {}}},{"type": "function","function": {"name": "get_ip_details","description": "获取IP详细信息（含城市、运营商）","parameters": {"type": "object","properties": {"ip": {"type": "string","description": "要查询的IP地址"}},"required": ["ip"]}}}, {"type": "function","function": {"name": "send_verification_code","description": "发送短信验证码","parameters": {"type": "object","properties": {"phone_number": {"type": "string","description": "手机号"},"code": {"type": "string","description": "验证码（需为6位数字）","pattern": "^\\d{6}$","minLength": 6,"maxLength": 6}},"required": ["phone_number","code"]}}},{"type": "function","function": {"name": "sent_email","description": "发送电子邮件","parameters": {"type": "object","properties": {"to": {"type": "string","description": "收件人邮箱地址"},"subject": {"type": "string","description": "邮件主题"},"body": {"type": "string","description": "邮件正文"}},"required": ["to","subject","body"]}}}
]

四、测试

为了测试数据操作，在本地创建了一个sqlite数据库，并创建了两个表students和scores，用来存储学生的个人信息和成绩，填充了一些假数据。

students表

scores表

1.测试数据库接口

读取表结构

select操作

提问和回答如下，获得的数据完全正确。

看一下response1.json内容，如下所示，可以看到function calling的响应中，大模型建议调用execute_query函数，并给出了正确的sql语句。

{"id": "chatcmpl-e0d887bb-3954-9585-9594-ae39ccbe5f75","choices": [{"finish_reason": "tool_calls","index": 0,"logprobs": null,"message": {"content": "","refusal": null,"role": "assistant","audio": null,"function_call": null,"tool_calls": [{"id": "call_ca7257e92b9a412d8c1553","function": {"arguments": "{\"sql\": \"SELECT * FROM scores WHERE 学号 = (SELECT 学号 FROM students WHERE 姓名 = ?)\", \"params\": [\"张晓萌\"]}","name": "execute_query"},"type": "function","index": 0}],"reasoning_content": ""}}],"created": 1747480960,"model": "qwen3-32b","object": "chat.completion","service_tier": null,"system_fingerprint": null,"usage": {"completion_tokens": 49,"prompt_tokens": 1319,"total_tokens": 1368,"completion_tokens_details": null,"prompt_tokens_details": null}
}