MCP 和 Function Calling：示例
本文以實際例子來加深對 MCP 和 Function Calling 的理解。實現(xiàn)這樣一個場景：和大模型聊天，然后讓大模型將回答的內(nèi)容總結(jié)后保存到 flomo 筆記中。
本文以實際例子來加深對 MCP 和 Function Calling 的理解。
實現(xiàn)這樣一個場景：和大模型聊天，然后讓大模型將回答的內(nèi)容總結(jié)后保存到 flomo 筆記中。
Function Calling
我們知道 Function Calling 和模型的能力有關，我使用的是 qwen2.5:7b 模型，用 ollama 在本地運行。
思路
- 寫一個 api 接口，這個接口的作用將輸入的內(nèi)容存入 flomo 筆記中。
- 利用 qwen-agent 框架來實現(xiàn) function calling ，最終調(diào)用自定義開發(fā)的 api 接口。
實現(xiàn)
api 接口使用任何語言都行，我這里使用的是 python 的 flask 框架。
@api_bp.route('/flomo/save', methods=['POST'])
def save_to_flomo():
# 獲取請求數(shù)據(jù)
data = request.get_json()
# 驗證請求數(shù)據(jù)
if not data or 'content' not in data:
return jsonify({"error": "Missing required field: content"}), 400
content = data['content']
tags = data.get('tags', []) # 可選的標簽列表
# 驗證Flomo API URL是否配置
flomo_api_url = current_app.config.get('FLOMO_API_URL')
if not flomo_api_url:
return jsonify({"error": "Flomo API URL not configured"}), 500
try:
# 準備發(fā)送到Flomo的數(shù)據(jù)
flomo_data = {
"content": content
}
# 如果有標簽，添加到內(nèi)容中
if tags:
# Flomo使用 #tag 格式的標簽
tag_text = ' '.join([f"#{tag}" for tag in tags])
flomo_data["content"] = f"{content}\n\n{tag_text}"
# 發(fā)送請求到Flomo API
headers = {
'Content-Type': 'application/json'
}
response = requests.post(
flomo_api_url,
headers=headers,
data=json.dumps(flomo_data),
timeout=10 # 設置超時時間，處理大文本可能需要更長時間
)
# 檢查響應
if response.status_code == 200:
return jsonify({
"message": "Content successfully saved to Flomo",
"flomo_response": response.json()
}), 200
else:
return jsonify({
"error": "Failed to save to Flomo",
"status_code": response.status_code,
"response": response.text
}), 500
except requests.RequestException as e:
# 處理請求異常
return jsonify({
"error": f"Request to Flomo failed: {str(e)}"
}), 500
except Exception as e:
# 處理其他異常
return jsonify({
"error": f"Unexpected error: {str(e)}"
}), 500創(chuàng)建一個 qwen-client.py 的文件，內(nèi)容如下：
import json
import requests
from qwen_agent.llm import get_chat_model
def save_to_flomo(content):
"""Save content to Flomo notes"""
try:
api_url = "http://localhost:6500/api/flomo/save"
data = {"content": content}
response = requests.post(
api_url,
headers={"Content-Type": "application/json"},
json=data,
timeout=10
)
if response.status_code == 200:
print(f"Successfully saved to Flomo: {content}")
return json.dumps(response.json())
else:
error_message = f"Failed to save to Flomo. Status code: {response.status_code}, Response: {response.text}"
print(error_message)
return json.dumps({"error": error_message})
except Exception as e:
error_message = f"Error calling Flomo API: {str(e)}"
print(error_message)
return json.dumps({"error": error_message})
def test(fncall_prompt_type: str = 'qwen'):
llm = get_chat_model({
'model': 'qwen2.5:7b',
'model_server': 'http://localhost:11434/v1',
'api_key': "",
'generate_cfg': {
'fncall_prompt_type': fncall_prompt_type
}
})
# 第1步：將對話和可用函數(shù)發(fā)送給模型
messages = [{'role': 'user', 'content': "怎么學習軟件架構(gòu)，總結(jié)為三點，保存到筆記"}]
functions = [{
'name': 'save_to_flomo',
'description': '保存內(nèi)容到Flomo筆記',
'parameters': {
'type': 'object',
'properties': {
'content': {
'type': 'string',
'description': '內(nèi)容',
}
},
'required': ['content'],
},
}]
responses = []
for responses in llm.chat(
messages=messages,
functions=functions,
stream=False,
):
print(responses)
# 如果使用stream=False，responses直接是結(jié)果，不需要循環(huán)
if isinstance(responses, list):
messages.extend(responses)
else:
messages.append(responses)
# 第2步：檢查模型是否想要調(diào)用函數(shù)
last_response = messages[-1]
if last_response.get('function_call', None):
# 第3步：調(diào)用函數(shù)
available_functions = {
'save_to_flomo': save_to_flomo,
}
function_name = last_response['function_call']['name']
function_to_call = available_functions[function_name]
function_args = json.loads(last_response['function_call']['arguments'])
function_response = function_to_call(
content=function_args.get('content'),
)
print('# Function Response:')
print(function_response)
# 第4步：發(fā)送每個函數(shù)調(diào)用和函數(shù)響應到模型，讓大模型返回最終的結(jié)果
messages.append({
'role': 'function',
'name': function_name,
'content': function_response,
})
for responses in llm.chat(
messages=messages,
functions=functions,
stream=False,
):
print(responses)
if __name__ == '__main__':
test()- save_to_flomo 方法就是大模型需要用到的函數(shù)，函數(shù)中調(diào)用第一步寫的接口，將內(nèi)容存儲到 flomo 筆記中。
- test 方法中首先進行初始化，http://localhost:11434/v1 是本地通過 ollama 運行 qwen2.5:7b 模型的地址。
- 后面的步驟在上面代碼中寫有注釋。
在 qwen_client.py 所在目錄執(zhí)行下面的命令安裝 qweb-agent 框架：
pip install -U "qwen-agent[gui,rag,code_interpreter,mcp]"執(zhí)行 python qwen_client.py 運行程序。
檢查 flomo 客戶端，可以看到內(nèi)容已經(jīng)存儲進來了。
MCP
MCP 的使用，可以自己開發(fā)服務端，也可以使用 MCP 服務站的服務，比如 mcp.so 。客戶端有很多，比如：Windsurf、Cursor、CherryStudio 等。
Windsurf 中使用 MCP
先在 mcp.so 中找到 flomo 的 Server 。
連接 Server 的方式選擇了 Original 。
在 Windsurf 中的 MCP 設置中添加 flomo 的 Server 。
配置好后，在 chat 模式下進行提問：“根據(jù)最新的內(nèi)容對比下 mcp 和 A2A，將結(jié)果存儲到筆記中”。
Windsurf 一通查詢，整理后，調(diào)用 MCP 工具，將結(jié)果存到我的 flomo 中了。
代碼示例
很久沒用 dotnet 了，這個例子就用 dotnet 來實現(xiàn)吧。
工具和環(huán)境：
- dotnet：8.0
- ModelContextProtocol：0.1.0-preview.8
- 工具：Windsurf
創(chuàng)建 mcp-server 控制臺項目，Program 代碼如下：
using Microsoft.Extensions.Hosting;
using ModelContextProtocol;
using Microsoft.Extensions.DependencyInjection;
using FlomoMcpServer;
try
{
Console.WriteLine("啟動 MCP 服務...");
var builder = Host.CreateEmptyApplicationBuilder(settings: null);
builder.Services
.AddMcpServer()
.WithStdioServerTransport()
.WithToolsFromAssembly();
await builder.Build().RunAsync();
}
catch (Exception)
{
Console.WriteLine("啟動 MCP 服務失敗");
}添加 flomo 工具類 FlomoTools.cs ，內(nèi)容如下：
using ModelContextProtocol.Server;
using System.ComponentModel;
namespace FlomoMcpServer
{
[McpServerToolType]
public static class FlomoTools
{
[McpServerTool]
[Description("寫筆記到 Flomo")]
public static async Task WriteNote(string content)
{
Console.WriteLine("寫筆記到 Flomo...");
if (string.IsNullOrEmpty(content))
{
throw new ArgumentNullException("content");
}
var apiUrl = "https://flomoapp.com/iwh/xxxxxxxxxxxxx/";
using (var httpClient = new HttpClient())
{
var payload = new { content = content };
var json = System.Text.Json.JsonSerializer.Serialize(payload);
var httpContent = new StringContent(json, System.Text.Encoding.UTF8, "application/json");
var response = await httpClient.PostAsync(apiUrl, httpContent);
response.EnsureSuccessStatusCode();
}
Console.WriteLine("寫筆記到 Flomo 完成");
}
}
}創(chuàng)建 mcp-client 控制臺項目，Program 代碼如下：
using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol.Transport;
using System.Collections.Generic;
var clientTransport = new StdioClientTransport(new StdioClientTransportOptions
{
Name = "flomo",
Command = "dotnet",
Arguments = new[] { "/Users/fengwei/Projects/ai-demo/dotnet-mcp-demo/mcp-server/bin/Debug/net8.0/mcp-server.dll" }
});
await using var client = await McpClientFactory.CreateAsync(clientTransport);
var tools = await client.ListToolsAsync();
foreach (var tool in tools)
{
Console.WriteLine($"{tool.Name}: {tool.Description}");
}上面例子中使用的是本地 Stdio 的模式。通過 client.ListToolsAsync(); 獲取 MCP 服務中的所有工具，并打印出來。執(zhí)行效果如下：
client 的 Program 中繼續(xù)添加下面代碼進行直接的 Server 端方法調(diào)用，來測試下 client 和 server 是否是連通的。
var result = await client.CallToolAsync("WriteNote", new Dictionary<string, object?>
{
["content"] = "Hello, oec2003!"
});
Console.WriteLine($"Result: {result}");執(zhí)行完后，如果 flomo 中筆記插入正常，說明調(diào)用成功。
接著調(diào)用本地 ollama 運行的大模型來實現(xiàn)跟大模型對話，然后將對話結(jié)果保存到 flomo 。Client 端的 Program 完整代碼如下：
using Microsoft.Extensions.Hosting;
using ModelContextProtocol;
using ModelContextProtocol.Client;
using ModelContextProtocol.Protocol.Transport;
using Microsoft.Extensions.DependencyInjection;
using System.Text;
using System.Text.Json;
using System.Net.Http;
using System.Net.Http.Json;
using Microsoft.Extensions.AI;
using OpenAI;
using System.ClientModel;
Console.WriteLine("啟動 MCP 客戶端...");
var clientTransport = new StdioClientTransport(new StdioClientTransportOptions
{
Name = "flomo",
Command = "dotnet",
Arguments = new[] { "/Users/fengwei/Projects/ai-demo/dotnet-mcp-demo/mcp-server/bin/Debug/net8.0/mcp-server.dll" }
});
await using var mcpClient = await McpClientFactory.CreateAsync(clientTransport);
Console.WriteLine("已連接到 MCP 服務器");
Console.WriteLine("可用工具:");
foreach (var tool in await mcpClient.ListToolsAsync())
{
Console.WriteLine($"{tool.Name}: {tool.Description}");
}
// 配置硅基流動API參數(shù)
var apiKeyCredential = new ApiKeyCredential("xx");
var aiClientOptions = new OpenAIClientOptions();
aiClientOptions.Endpoint = new Uri("http://localhost:11434/v1");
var aiClient = new OpenAIClient(apiKeyCredential, aiClientOptions)
.AsChatClient("qwen2.5:7b");
var chatClient = new ChatClientBuilder(aiClient)
.UseFunctionInvocation()
.Build();
var mcpTools = await mcpClient.ListToolsAsync();
var chatOptions = new ChatOptions() {
Tools = [..mcpTools]
};
Console.ForegroundColor = ConsoleColor.Green;
Console.WriteLine($"助手> 請輸入想要記錄的內(nèi)容，AI總結(jié)后會存入筆記");
while (true)
{
Console.ForegroundColor = ConsoleColor.White;
Console.Write("用戶> ");
var question = Console.ReadLine();
if (!string.IsNullOrWhiteSpace(question) && question.ToUpper() == "EXIT")
break;
var messages = new List<ChatMessage> {
new(ChatRole.System, "你是一個筆記助手，請將用戶的輸入總結(jié)為簡潔的筆記形式，使用markdown格式。保留關鍵信息，刪除冗余內(nèi)容。"),
new(ChatRole.User, question)
};
try
{
var response = await chatClient.GetResponseAsync(messages, chatOptions);
var content = response.ToString();
Console.WriteLine($"助手> {content}");
}
catch (Exception ex)
{
Console.WriteLine($"錯誤: {ex.Message}");
}
Console.WriteLine();
}輸入 dotnet run 運行程序，結(jié)果如下：
責任編輯：姜華
來源：
不止dotNET
