# Function Calling(函数调用)
Function Calling 允许大语言模型与外部工具连接，将自然语言转换为API调用。
# Function Calling的工作原理：
- 函数定义：开发者定义函数的名称、描述、参数及其类型
- 模型推理：分析用户请求，判断是否需要调用函数
- 参数生成：生成函数所需的参数，以JSON格式输出
- 函数执行：开发者根据模型输出执行实际函数
- 结果整合： 将函数执行结果返回给模型，生成最终答案
# 局限性
每个LLM供应商的接口格式有差异，开发者想支持多个模型，需要为不同API适配或使用额外的框架处理。而且无状态性：模型仅用于调用规范，实际执行还是由外部系统完成。
# MCP协议与开发
MCP（Model Content Protocol,模型上下文协议），旨在统一大型语言模型（LLM）与外部数据源和工具之间的通信协议。MCP的主要目的在于解决当前AI模型因数据孤岛而无法充分发挥潜力的难题，MCP是的AI应用能够安全的访问和操作本地及远程数据，为AI应用提供了连接万物的接口。
Function Calling是AI模型调用函数的机制，MCP是一个标准协议，使AI模型与API无缝交互，而AI Agent是一个自主运行的智能系统，利用Function Calling和MCP来解析和执行任务，实现特定目标。
MCP可以直接在AI与数据之间架起一座桥梁，通过MCP服务器和MCP客户端，大家只要遵循这个协议，就能实现“万物互联”。
# 工作原理
MCP协议采用了一种独特的架构设计，它将LLM与资源之间的通信划分为三个主要部分:客户端、服务器和资源。客户端负责发送请求给MCP服务器，服务器则将这些请求转发给相应的资源。这种分层的设计使得MCP协议能够更好地控制访问权限，确保只有经过授权的用户才能访问特定的资源
以下是 MCP 的基本工作流程:
- 初始化连接:客户端向服务器发送连接请求，建立通信通道。
- 发送请求:客户端根据需求构建请求消息，并发送给服务器。
- 处理请求:服务器接收到请求后，解析请求内容，执行相应的操作(如查询数据库、读取文件等)
- 返回结果:服务器将处理结果封装成响应消息，发送回客户端。
- 断开连接:任务完成后，客户端可以主动关闭连接或等待服务器超时关闭。
# 通信机制
1、基于标准输入和输出的本地通信（stdio）
2、基于SSE（Server-Sent Events）的远程通信。是一种基于HTTP协议的单向通信协议，允许服务器以事件流的形式实时向客户端推送数据，而无需客户端明确请求。MCP中的SSE Transport 结合了SSE技术和HTTP POST
HTTP+SSE的缺陷
- 不支持恢复连接
- 要求服务器保持高可用的长连接
- 服务器只能通过SSE发送消息（服务器无法在已有的请求之外，主动地发送消息给客户端，除了通过专门的sse通道。“单向被动响应，而不是任意时机推送”）
3、streamable HTTP
streamable HTTP是MCP协议在2025年3月引入的一种新传输机制，旨在取代之前的HTTP+SSE传输模式。它的设计理念是在保留SSE优点的同事克服其限制。
Streamable HTTP的核心思想是提供一个统一的HTTP断电，同时支持POST和GET方法：
a、POST方法：用于客户端向服务端发送请求和接收响应
b、GET方法（可选）：用于建立SSE流，接收服务器实时推送的消息。
1.连接:在streamable HTTP模式下，并没有和SSE模式类似的"连接“过程(在sse_client调用时)，因为无需事先创建SSE连接;
2.客户端发起初始化请求(Initialize)。如果是有状态模式，会在返回消息的HTTP头中携带session-id;
3.客户端发起初始化确认(Initialized)。此时如果已有session-id(有状态)，客户端会首先发起一次HTTP Get请求，以建立独立的SSE通道；
4.后续正常交互:普通的交互都是通过Post通道来进行，只有两种情况会使用SSE通道:服务端发起的通知与请求、以及会话恢复的事件发送。
# FastMCP:构建模型上下文协议（MCP）服务器的快速方案
# 开发mcp服务器
# python版本
使用fastmcp库
pip install fastmcp
mcp工具、提示词、资源
from fastmcp import FastMCP
import aiohttp
mcp_server = FastMCP(name="vitali", instructions="vitali的MCP")
@mcp_server.tool(name="web_search") # name不定义，就是函数名字
async def web_search(query: str) -> str:
"""搜索互联网上的内容"""
url = 'https://www.searchapi.io/api/v1/search'
params = {
'api_key': 'TYYv2iVR37', # 错误的
'engine': 'google',
'q': query
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status != 200:
return f"API 请求失败，状态码: {response.status}"
result = await response.text()
return result
except Exception as e:
return f"网络请求发生异常: {str(e)}"
@mcp_server.tool(description="给指定的用户打招呼")
async def say_hello(username: str) -> str:
"""给指定的用户打招呼"""
return f"你好,{username},天天开心"
# prompt模板,返回结构化消息对象
from fastmcp.prompts.prompt import PromptMessage, TextContent
@mcp_server.prompt()
def gen_code_request(language: str, task_description: str) -> PromptMessage:
"""生成代码编写请求的提示词模板"""
content = f"请用{language}编写代码实现以下功能：\n{task_description}"
return PromptMessage(
role='user',
content=TextContent(type='text', text=content)
)
# 结构化资源
@mcp_server.resource("resource://config")
def get_config() -> dict:
"""以json格式返回应用配置"""
return {
'theme': 'Dark',
'version': '1.2.1',
'features': ['tools', 'resources']
}sse通信机制启动mcp
from tool_server import mcp_server
if __name__ == "__main__":
mcp_server.run(transport='sse', host='127.0.0.1',port=8080, log_level='debug', path='/sse')- transport:传输协议类型
- host：监听的ip地址
- port：监听的端口号
- log_level：日志级别
- path：SSE服务的URL路由
- 实际访问地址
http://127.0.0.1:8080/sse
- 实际访问地址
streamable通信机制启动mcp
from tool_server import mcp_server
if __name__ == "__main__":
mcp_server.run(transport='streamable-http', host='127.0.0.1',port=8080, log_level='debug', path='/streamable') # path默认是 /mcp
# mcp客户端（调用mcp的智能体）
使用库
pip install langchain-mcp-adapters调用上面自己写的mcp服务
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from dotenv import load_dotenv
import os
load_dotenv(r'D:\Desktop\LLM-LE\.env')
# python MCP服务端配置
python_mcp_server_config = {
# 'url': 'http://127.0.0.1:8080/sse',
# 'transport': 'sse',
'url': 'http://127.0.0.1:8080/streamable',
'transport': 'streamable_http'
}
# mcp客户端
mcp_client = MultiServerMCPClient(
{
'python_mcp': python_mcp_server_config,
}
)
llm = ChatOpenAI(
base_url=os.getenv('OPENAI_BASE_URL'),
api_key=os.getenv('OPENAI_API_KEY'),
model=os.getenv('OPENAI_MODEL'),
)
async def create_agent():
'''必须是异步函数中'''
# 获取mcp工具
mcp_tools = await mcp_client.get_tools() # 不加参数就是把所有的工具都拿出来
# 获取mcp prompt server_name和prompt_name 是必传的，arguments是提示词模板需要的参数 一次只能拿一个
p = await mcp_client.get_prompt(server_name='python_mcp', prompt_name='gen_code_request', arguments={'language': 'python', 'task_description': '输出你好'})
# 获取资源
data = await mcp_client.get_resources(server_name='python_mcp',uris='resource://config')
print(f'tools--{mcp_tools}',
f'prompt--{p}'
f'resource--{data[0]}'
)
# 创建agent时传入工具
return create_react_agent(
model=llm,
tools=mcp_tools,
prompt="你是一个智能助手，尽可能调用工具回答用户问题"
)
import asyncio
agent = asyncio.run(create_agent())
result = agent.invoke({"messages": [{"role": "user", "content": "马斯克有几家公司"}]})
print(result)
"""
可以看到
tools--[StructuredTool(name='web_search', description='搜索互联网上的内容', args_schema={'properties': {'query': {'type': 'string'}}, 'required': ['query'], 'type': 'object'}, metadata={'_meta': {'_fastmcp': {'tags': []}}}, response_format='content_and_artifact', coroutine=<function convert_mcp_tool_to_langchain_tool.<locals>.call_tool at 0x000002DB042DFEC0>), StructuredTool(name='say_hello', description='给指定的用户打招呼', args_schema={'properties': {'username': {'type': 'string'}}, 'required': ['username'], 'type': 'object'}, metadata={'_meta': {'_fastmcp': {'tags': []}}}, response_format='content_and_artifact', coroutine=<function convert_mcp_tool_to_langchain_tool.<locals>.call_tool at 0x000002DB042DDB20>)]
prompt--[HumanMessage(content='请用python编写代码实现以下功能：\n输出你好', additional_kwargs={}, response_metadata={})]
resource--metadata={'uri': 'resource://config'} data='{"theme":"Dark","version":"1.2.1","features":["tools","resources"]}' mimetype='text/plain'
"""
"""
{'messages': [HumanMessage(content='给张三打个招呼', additional_kwargs={}, response_metadata={}, id='06301854-a9e0-4843-a13f-608ac066e28a'), AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 16, 'prompt_tokens': 89, 'total_tokens': 105, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-4.1-mini-2025-04-14', 'system_fingerprint': 'fp_2191215734', 'id': 'chatcmpl-D0p4Jmq9b3CkcOicwxE5ud0meQSI1', 'service_tier': 'default', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019be5e6-f719-7301-b5bc-ddccab42b87a-0', tool_calls=[{'name': 'say_hello', 'args': {'username': '张三'}, 'id': 'call_YoSiNULvtYbbV0NgnR4o2wHQ', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 89, 'output_tokens': 16, 'total_tokens': 105, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}), ToolMessage(content=[{'type': 'text', 'text': '你好,张三,天天开心', 'id': 'lc_0fd7f61e-ae79-410a-a141-8b2dc27b1e7f'}], name='say_hello', id='1f98303f-76e4-4df9-a49a-7c7fb8308f76', tool_call_id='call_YoSiNULvtYbbV0NgnR4o2wHQ', artifact={'structured_content': {'result': '你好,张三,天天开心'}}), AIMessage(content='我已经给张三打了招呼，内容是：你好,张三,天天开心。 还有什么需要帮忙的吗？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 28, 'prompt_tokens': 120, 'total_tokens': 148, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-4.1-mini-2025-04-14', 'system_fingerprint': 'fp_376a7ccef1', 'id': 'chatcmpl-D0p4LhFNYd8Gr1puMfjuaN4lLHXEB', 'service_tier': 'default', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019be5e7-02b1-7231-a727-bdc43ab40ae9-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 120, 'output_tokens': 28, 'total_tokens': 148, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})]}
"""调用外网的mcp
zhipu_mcp_server_config = {
'url': 'https://open.bigmodel.cn/api/mcp/web_search_prime/sse?Authorization=your_api_key',
'transport': 'sse'
}
# mcp客户端
mcp_client = MultiServerMCPClient(
{
# 'python_mcp': python_mcp_server_config,
'zhipu_mcp': zhipu_mcp_server_config
}
)
...
"""
可以看到有很多工具
tools--[StructuredTool(name='webSearchPrime', description='Search web information, returns results including web page title, web page URL, web page summary, website name, website icon, etc.', args_schema={'type': 'object', 'properties': {'search_query': {'type': 'string', 'description': 'Content to be searched, it is recommended that the search query not exceed 70 characters'}, 'search_domain_filter': {'type': 'string', 'description': 'Used to limit the scope of search results, only return content from specified whitelist domains, such as: www.example.com.'}, 'search_recency_filter': {'type': 'string', 'description': 'Search for web pages within a specified time range. Default is noLimit Available values:- oneDay, within one day- oneWeek, within one week- oneMonth, within one month- oneYear, within one year- noLimit, no limit (default)'}, 'content_size': {'type': 'string', 'description': 'Control the number of words in the web page summary; default value is medium - medium: balanced mode, suitable for most queries. 400-600 words.- high: maximize context to provide more comprehensive answers, but higher cost, 2500 words.'}, 'location': {'type': 'string', 'description': 'Guess which region the user is from based on user input. Default is cn Available values: - cn, Chinese region- us, non-Chinese region'}}, 'required': ['search_query'], 'additionalProperties': False}, response_format='content_and_artifact', coroutine=<function convert_mcp_tool_to_langchain_tool.<locals>.call_tool at 0x000001F9FF1E1760>)]
"""