DeepSeek+dify知識庫，查詢數(shù)據(jù)庫的兩種方式（api+直連）
MCP 協(xié)議由一個 client 和多個 server 組成。用戶發(fā)出請求后，client 會去調(diào)用 server 執(zhí)行邏輯并返回給用戶。
client 就是用戶的對話框，server 是一些我們提前寫好的代碼，用來完成我們想實現(xiàn)的系統(tǒng)調(diào)用。client 和 server 都可以自行開發(fā)。
工具是服務(wù)器編寫，可由客戶端調(diào)用并由 LLM 用來執(zhí)行操作的可執(zhí)行函數(shù)。
當前對其支持較多，主要的邏輯都通過工具來實現(xiàn)。
可以理解為預設(shè)好的提示詞模版，用于補充用戶的輸入。
假設(shè)用戶輸入的是：
def add(a, b):
return a + b生成的提示可能是：
Explain how this Python code works:
def add(a, b):
return a + b當前對其的支持相對較少。
提供服務(wù)端的資源給客戶端。
當前對其支持相對較少，一般只有客戶端顯示調(diào)用的時候才會使用。
MCP 的工作流程可以在這里找到：https://github.com/modelcontextprotocol/python-sdk/tree/main/examples/clients/simple-chatbot
簡單總結(jié)如下：
官方 demo 中的提示詞如下
system_message = (
"You are a helpful assistant with access to these tools:\n\n"
f"{tools_description}\n"
"Choose the appropriate tool based on the user's question. "
"If no tool is needed, reply directly.\n\n"
"IMPORTANT: When you need to use a tool, you must ONLY respond with "
"the exact JSON object format below, nothing else:\n"
"{\n"
' "tool": "tool-name",\n'
' "arguments": {\n'
' "argument-name": "value"\n'
" }\n"
"}\n\n"
"After receiving a tool's response:\n"
"1. Transform the raw data into a natural, conversational response\n"
"2. Keep responses concise but informative\n"
"3. Focus on the most relevant information\n"
"4. Use appropriate context from the user's question\n"
"5. Avoid simply repeating the raw data\n\n"
"Please use only the tools that are explicitly defined above."
)使用 MCP，我們需要：
下面使用 Claude模型，使用 Claude Desktop 作為 MCP Client，自己開發(fā) MCP Server 進行使用。(也可以直接使用官方的 MCP Server)
使用 python sdk 開發(fā)，前置需要 Mac，并安裝 python uv 和 Claudedesktop。
官網(wǎng)給出的開發(fā)流程：https://modelcontextprotocol.io/quickstart/server
使用 uv 創(chuàng)建開發(fā)環(huán)境
# Create a new directory for our project
uv init 項目名
cd 項目名
# Create virtual environment and activate it
uv venv
source .venv/bin/activate
# Install dependencies
uv add "mcp[cli]" httpx官網(wǎng)上有 demo (https://modelcontextprotocol.io/quickstart/server) 可以直接丟給大模型仿照這個寫。
寫提示詞的時候要注意下 mcp server 的提示詞來源，閱讀源代碼發(fā)現(xiàn)它有兩種方式：用戶手動指定 description，就是自己寫提示詞，如果用戶不手動指定的話，就默認使用 mcp server 代碼的注釋作為提示詞。
編寫完代碼，測試能成功運行之后，需要修改 claude desktop 的配置使其識別到這個 server，配置路徑：
~/Library/Application\ Support/Claude/claude_desktop_config.json配置的格式
{
"mcpServers": {
"weather": {
"command": "uv",
"args": [
"--directory",
"/Users/wpy/downloads/mcp-test",
"run",
"hello.py"
]
}
}
}保存并退出之后，重啟 Claudedesktop，就能看到對話框的右下角出現(xiàn)一個錘子標志：
出現(xiàn)這個標志就表明 desktop 已經(jīng)識別到了編寫的 server，接下來就可以開始使用了。
針對提出的問題，模型會自主判斷是否要使用這些 server，并在使用的時候發(fā)出提示：
MCP Server 運行在本地，直接訪問本地的文件。
server 代碼：
import os
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("file")
@mcp.tool()
async def list_directory_contents(directory: str) -> str:
"""List the contents of a directory.
Args:
directory: The path of the directory to list.
"""
try:
# Get the list of files and directories in the specified path
directory_contents = os.listdir(directory)
if not directory_contents:
return "The directory is empty."
# Format the contents into a readable list
contents = "\n".join(directory_contents)
return f"Contents of {directory}:\n{contents}"
except FileNotFoundError:
return f"Directory {directory} not found."
except PermissionError:
return f"Permission denied to access {directory}."
except Exception as e:
return f"An error occurred: {str(e)}"
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')Claude desktop 的配置文件：
{
"file": {
"command": "uv",
"args": [
"--directory",
"/Users/wpy/downloads/file-mcp",
"run",
"hello.py"
]
}
}
}可以查看本地目錄下的文件。
本地運行 vlogs server，通過 MCP 的 Server 去訪問。
server 代碼：
import os
import requests
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("file")
# Define the URL for the API endpoint
API_URL = "http://localhost:8428/queryLogsByParams"
HEADERS = {
"Content-Type": "application/json",
"Authorization": "Bearer my-token"
}
@mcp.tool()
async def get_logs(time: str, namespace: str, app: str, limit: str, pod: list, container: list, keyword: str) -> str:
"""Get logs from the API by passing parameters.
Args:
time: The time filter for logs (e.g., "10h").
namespace: The namespace of the application (e.g., "gpu-operator").
app: The name of the application (e.g., "gpu-operator").
limit: The limit for the number of logs to retrieve.
pod: The list of pods to query.
container: The list of containers to query.
keyword: The keyword to filter logs by.
"""
data = {
"time": time,
"namespace": namespace,
"app": app,
"limit": limit,
"jsonMode": "false",
"stderrMode": "true",
"numberMode": "false",
"numberLevel": "h",
"pod": pod,
"container": container,
"keyword": keyword,
"jsonQuery": []
}
try:
# Make the POST request to the API
response = requests.post(API_URL, headers=HEADERS, json=data)
if response.status_code == 200:
# If the request was successful, return the log data
return f"Logs retrieved successfully:\n{response.text}"
else:
return f"Failed to retrieve logs. Status code: {response.status_code}"
except requests.exceptions.RequestException as e:
return f"An error occurred while accessing the API: {str(e)}"
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')Claude desktop 的配置文件：
{
"mcpServers": {
"vlogs-server": {
"command": "uv",
"args": [
"--directory",
"/Users/wpy/downloads/vlogs-mcp",
"run",
"hello.py"
]
}
}
}可以實現(xiàn)使用 AI 查詢并分析用戶的日志。
從效果上看，可以滿足我們的部分需求。
從效果上看，像上面的 demo 一樣，開發(fā)一套 MCP Server 和 Client 可以滿足我們的部分需求，實現(xiàn) AI 的工程化接入。
能實現(xiàn)的需求包括：
還可以根據(jù)其官方和社區(qū)提供的各種 MCP Server 思考新的功能
之前的 demo 中 MCP Server 的開發(fā)都依賴了官方的 SDK，但是官方的 SDK 存在一些問題
可以通過一些操作解決之前提到的問題，而不重寫 SDK。
總結(jié)：本地 AI-Coding 就在 cline 基礎(chǔ)上內(nèi)置一些 MCP 和環(huán)境信息，Web 網(wǎng)站上就在 Nextjs 端做一下 MCP 即可。
Claude 的 SDK 都開源了 (例如 python-sdk 位于 https://github.com/modelcontextprotocol/python-sdk)，可以仿照開發(fā)出其他語言的 SDK。
通過重寫 SDK，可以把 MCP Server 放在遠程并支持其他模型，用戶本地不需要安裝一堆的 server，除了一個 client，其他都放在遠程：
采用這個方案，成本更高，但是可以完全符合我們的需求，并且可以考慮后續(xù)和其官方社區(qū)合作。
MCP 協(xié)議稍微有點復雜，不過其中的一些設(shè)計理念可以參考。
本文轉(zhuǎn)載自公眾號@sealos
鍵.png)
