字数 2925，浏览约莫需 15 分钟
原文主要介绍怎样完毕一个鉴于DeepSeek的MCP Client。
那个Client能够用去盘问并前去网页实质、归纳谈天疑息并保存到当地文献，大概把网页盘问的疑息间接保留到当地。
那里咱们会用到二个MCP server去帮助展示Client的结果:
1. fetch : 一个特地用于获得网页实质的MCP效劳器2. filesystem 用于当地文献体系操纵的MCP效劳器
您也能够散成其余MCP Server, 与决于需要完毕甚么样的事情流。
简介
Claude公布MCP已经有一段时间了，近来不竭正在分离各类编程硬件战MCP server去帮助编程、阐发数据战写代码。
微疑公家号上瞅了很多写MCP观点、道理战使用的文章，整体来讲各类理解皆有，年夜大都最初皆是用cursor,cline, windsurf等挪用一下某个server完事。MCP server的完毕也确实很简朴，可是尔正在念，正在企业真实降天大要率仍是要自己来完毕Client, 而后来调整agents和各种server完毕一定的营业逻辑。
假设您已经瞅了吴恩达DeepLearning.ai上对于MCP的系列课程:mcp-build-rich-context-ai-apps-with-anthropic[1]，念必对于MCP的道理，完毕战Claude未来的计划有了必然的理解。假设 尚未瞅过，剧烈倡议来瞅一下，赛过99%的公账号。
那个课程最年夜的限定是使用Claude的API去完毕MCP Client,关于 海内用户来讲可操纵性没有强，以是尔写了一个鉴于DeepSeek的MCP客户端，用去帮助减深对于MCP架构的理解。
MCP Client Server接互逻辑
先去理解下MCP Client战Server是怎样接互的：
【初初化阶段】
1. 【Client】剖析servers.json文献，获得mcp servers的设置。2. 【Client】使用MCP供给的ClientSession取统统设置的mcp server成立跟尾。3. 【Client】读与每一个server的tools，将其变换成适配OpenAI的function格局，保存到列表中。
【接互阶段】
1. 【用户】输出prompt, 提出成就。2. 【Client】将用户的prompt战tools列表共同传给LLM。3. 【LLM】鉴别可否需要挪用mcp tool,假设 没有需要，则间接前去成果，假设需要，则前去tools列表。4. 【Client】假设需要挪用mcp tool, 则按照已经获得的tool列表，挪用对于应的mcp tool, 将成果前去给LLM。5. 【LLM】重复步调3-4，曲到没有需要挪用mcp tool为行。6. 【Client】将LLM的前去成果截至处置，并前去给用户。
名目实践
交下来咱们去完毕那个MCP Client
前置事情
1. 备案DeepSeek账号并获得API Key2.装置 python3.装置 uv4.装置 nodejs
创立名目
#创立 名目
uv init mcp-client
cd mcp-client
#创立 假造情况
uv venv
source .venv/bin/activate
#装置 依靠
uv add mcp nest-asyncio openai python-dotenv
#创立 client.py
touch client.py
#创立 下载文献夹
mkdir downloads
创立.env文献
touch .env
树立情况变质
DEEPSEEK_API_KEY=your_api_key
把.env文献增加到.gitignore文献中
echo ".env" >> .gitignore3. 正在名目根目次下增加servers.json， 实质以下：
{"mcpServers":{
"mcp_server_fetch":{
"co妹妹and":"uvx",
"args":["mcp-server-fetch"]
},
"filesystem":{
"co妹妹and":"npx",
"args":[
"-y",
"@modelcontextprotocol/server-filesystem",
"./downloads"
]
}
}
}代码完毕
先按一般的逻辑走一遍，咱们需要完毕：
1. 运行剧本时，需要完毕根底的对于话功用2. 使用MCPClient去截至初初化事情3. 正在MCPClient中完毕根底的对于话功用战资本清理事情
如下是据具体的代码完毕，代码为脚撸，正文使用LLM后增加，便利理解细节
1. 客户端初初化
class MCPClient:
def __init__(self):
self.max_loop = 10 #避免 无限轮回的最年夜对于话轮次限定
self.messages = [] # 保存残破的对于话汗青，用于连结高低文
self.exit_stack = AsyncExitStack() # 同步资本办理器，保证统统跟尾准确封闭
self.sessions: List[ClientSession] = [] # 保存统统MCP效劳器会话
self.available_tools: List[dict] = [] # 保存变换为OpenAI格局的东西列表
self.tool_session_mapping: Dict[str, ClientSession] = {} # 东西称呼到会话的映照，用于路由东西挪用
# 初初化DeepSeek客户端，从情况变质获得API稀钥
self.client = AsyncOpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")2. 跟尾单个MCP效劳器
async defconnect_to_server(self, server_name: str, server_config: dict):
"""跟尾单个MCP效劳器并获得其东西列表"""
try:
#依据 设置创立效劳器参数
server_params = StdioServerParameters(
co妹妹and=server_config["co妹妹and"], #效劳 器启用号令
args=server_config["args"], # 号令止参数
env=server_config.get("env"), # 情况变质（可选）
)
# 颠末stdio和谈成立取MCP效劳器的单背通信
stdio_transport = awaitself.exit_stack.enter_async_context(stdio_client(server_params))
stdio, write = stdio_transport
#创立 客户端会话
session = awaitself.exit_stack.enter_async_context(ClientSession(stdio, write))
self.sessions.append(session)
logger.debug(f"Connected to server: {server_name}")
# 初初化会话，成立和谈握脚
await session.initialize()
# 获得效劳器供给的东西列表
response = await session.list_tools()
logger.debug(f"Tool Response: {response}")
# 将MCP东西变换为OpenAI函数挪用格局
for tool in response.tools:
self.tool_session_mapping[tool.name] = session #树立 东西到会话的映照
self.available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema # MCP的inputSchema间接兼容OpenAI的parameters格局
}
})
except Exception as e:
logger.error(f"Error connecting to server {server_name}: {str(e)}")
raise3. 跟尾统统设置的效劳器
async defconnect_to_servers(self):
"""从servers.json读与设置并跟尾统统MCP效劳器"""
try:
# 读与效劳器设置文献
withopen("servers.json", "r") as f:
data = json.load(f)
servers = data.get("mcpServers", {})
# 遍历设置文献中的统统效劳器，一一成立跟尾
for server_name, server_config in servers.items():
awaitself.connect_to_server(server_name, server_config)
except FileNotFoundError:
logger.error("No servers.json file found")
raise
except Exception as e:
logger.error(f"Error connecting to servers: {str(e)}")
raise4.处置 用户盘问的中心逻辑
async defprocess_query(self, query: str, system_prompt: Optional[str] = None) -> None:
"""处置用户盘问，完毕LLM取东西挪用的残破轮回"""
#假设 有体系提醒词汇，先增加到消息汗青
if system_prompt:
self.messages.append({"role": "system", "content": system_prompt})
# 增加用户盘问到消息汗青
self.messages.append({"role": "user", "content": query})
# 开端LLM取东西挪用的轮回，至多施行max_loop次避免无限轮回
for i inrange(self.max_loop):
try:
# 挪用DeepSeek API，传进对于话汗青战可用功具列表
response = awaitself.client.chat.completions.create(
model="deepseek-chat",
messages=self.messages,
tools=self.available_tools, # 传进统统可用的MCP东西
tool_choice="auto" # 让AI主动鉴别可否需要挪用东西
)
logger.debug(f"Chat Completion Response: {response}")
except Exception as e:
logger.error(f"Error processing query: {str(e)}")
raise
# 获得AI的照应消息
response_message = response.choices[0].message
# 将AI照应增加到消息汗青
self.messages.append({
"role": "assistant",
"content": response_message.content,
"tool_calls": response_message.tool_calls
})
#假设 AI没有需要挪用东西，间接输出成果并完毕轮回
ifnot response_message.tool_calls:
print("AI: ", response_message.content)
break
# AI决定挪用东西，遍历统统东西挪用恳求
for tool_call in response_message.tool_calls:
tool_name = tool_call.function.name
# 剖析东西挪用参数（OpenAI前去JSON字符串，需要变换为字典）
try:
arguments = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
arguments = {}
print(f"Calling tool: [{tool_name}] with arguments: {arguments}")
try:
#依据 东西称呼找到对于应的MCP会话
session = self.tool_session_mapping[tool_name]
# 挪用MCP东西
mcp_result = await session.call_tool(
name=tool_name,
arguments=arguments
)
logger.debug(f"MCP Tool call Result: {mcp_result}")
#处置 东西挪用成果，变换为字符串格局
ifhasattr(mcp_result, 'content'):
ifisinstance(mcp_result.content, list):
result_content = "\n".join([str(item) for item in mcp_result.content])
else:
result_content = str(mcp_result.content)
else:
result_content = str(mcp_result)
# 将东西挪用成果增加到消息汗青，供AI下一轮使用
self.messages.append({
"role": "tool",
"name": tool_name,
"content": result_content,
"tool_call_id": tool_call.id
})
except Exception as e:
# 东西挪用失利时，将毛病疑息增加到消息汗青
error_msg = f"Error: {str(e)}"
logger.error(error_msg)
self.messages.append({
"role": "tool",
"name": tool_name,
"content": error_msg,
"tool_call_id": tool_call.id
})5. 接互式谈天轮回
async defchat_loop(self):
"""运行接互式谈天轮回，处置用户输出"""
print("\nMCP Client Started!")
print("Type 'exit' or 'quit' to quit.\n")
whileTrue:
try:
# 获得用户输出
query = input("\nYou: ").strip()
if query.lower() == "exit"or query.lower() == "quit":
break# 用户挑选参加
ifnot query:
continue#疏忽 空输出
# 调试情势下挨印消息汗青
ifself.messages:
logger.debug(f"Messages[{len(self.messages)}]:")
for message inself.messages:
logger.debug(message)
#处置 用户盘问
awaitself.process_query(query)
except Exception as e:
logger.error(f"Error: {str(e)}")6. 资本清理战法式进口
async defcleanup(self):
"""清理统统同步资本，保证跟尾准确封闭"""
awaitself.exit_stack.aclose()
asyncdefmain():
"""法式主进口，办理全部性命周期"""
client = MCPClient()
try:
# 初初化阶段：跟尾统统设置的MCP效劳器
await client.connect_to_servers()
# 接互阶段：启用谈天轮回
await client.chat_loop()
except KeyboardInterrupt:
print("Goodbye!")
finally:
# 保证资本准确开释
await client.cleanup()
if __name__ == "__main__":
asyncio.run(main())结果示范
先运行剧本：
python client.py
假设尔来日诰日念要理解一下uv号令如何使用，那末能够这样提问：
tell me how to use uv to manage python packages from the official documentation and save it as a markdown file
成果是如许的：
MCP Client Started!
Type 'exit' or 'quit' to quit.
You: tell me how to use uv to manage python packages from the official documentation and save it as a markdown file
Calling tool: [fetch] with arguments: {'url': 'https://pypi.org/project/uv/', 'max_length': 5000}
Calling tool: [fetch] with arguments: {'url': 'https://docs.astral.sh/uv/', 'max_length': 5000}
Calling tool: [fetch] with arguments: {'url': 'https://docs.astral.sh/uv/', 'start_index': 5000, 'max_length': 5000}
Calling tool: [write_file] with arguments: {'path': 'uv_package_management.md', 'content': '# Using `uv` to Manage Python Packages\n\n`uv` is an extremely fast Python package and project manager written in Rust. It aims to replace tools like `pip`, `pip-tools`, `pipx`, `poetry`, `pyenv`, `twine`, and `virtualenv` with a single, efficient tool.\n\n## Installation\n\nYou can install `uv` using the official standalone installer or via `pip`:\n\n### Standalone Installer\n\n**macOS and Linux:**\n```bash\ncurl -LsSf https://astral.sh/uv/install.sh | sh\n```\n\n**Windows:**\n```powershell\npowershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"\n```\n\n### Via `pip`\n```bash\npip install uv\n```\n\n### Via `pipx`\n```bash\npipx install uv\n```\n\nAfter installation, you can update `uv` itself:\n```bash\nuv self update\n```\n\n## Features\n\n### Projects\n\n`uv` manages project dependencies and environments, similar to `rye` or `poetry`:\n\n```bash\nuv init example\ncd example\nuv add ruff\nuv run ruff check\nuv lock\nuv sync\n```\n\n### Scripts\n\n`uv` can manage dependencies for single-file scripts with inline metadata:\n\n```bash\necho \'import requests; print(requests.get("https://astral.sh"))\' > example.py\nuv add --script example.py requests\nuv run example.py\n```\n\n### Tools\n\n`uv` can execute and install co妹妹and-line tools provided by Python packages, similar to `pipx`:\n\n```bash\nuvx pycowsay \'hello world!\'\nuv tool install ruff\n```\n\n### Python Versions\n\n`uv` can install and manage Python versions:\n\n```bash\nuv python install 3.10 3.11 3.12\nuv venv --python 3.12.0\nuv python pin 3.11\n```\n\n### The `pip` Interface\n\n`uv` provides a drop-in replacement for co妹妹on `pip`, `pip-tools`, and `virtualenv` co妹妹ands:\n\n```bash\nuv pip compile docs/requirements.in --universal --output-file docs/requirements.txt\nuv venv\nuv pip sync docs/requirements.txt\n```\n\n## Learn More\n\nFor detailed documentation, visit [docs.astral.sh/uv](https://docs.astral.sh/uv).\n\nYou can also explore the [first steps](https://docs.astral.sh/uv/getting-started/first-steps/) or jump straight to the [guides](https://docs.astral.sh/uv/guides/).'}
Calling tool: [list_allowed_directories] with arguments: {}
Calling tool: [write_file] with arguments: {'path': '/Users/loong_zhou/mcp/test-mcp/mcp_demo/client/downloads/uv_package_management.md', 'content': '# Using `uv` to Manage Python Packages\n\n`uv` is an extremely fast Python package and project manager written in Rust. It aims to replace tools like `pip`, `pip-tools`, `pipx`, `poetry`, `pyenv`, `twine`, and `virtualenv` with a single, efficient tool.\n\n## Installation\n\nYou can install `uv` using the official standalone installer or via `pip`:\n\n### Standalone Installer\n\n**macOS and Linux:**\n```bash\ncurl -LsSf https://astral.sh/uv/install.sh | sh\n```\n\n**Windows:**\n```powershell\npowershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"\n```\n\n### Via `pip`\n```bash\npip install uv\n```\n\n### Via `pipx`\n```bash\npipx install uv\n```\n\nAfter installation, you can update `uv` itself:\n```bash\nuv self update\n```\n\n## Features\n\n### Projects\n\n`uv` manages project dependencies and environments, similar to `rye` or `poetry`:\n\n```bash\nuv init example\ncd example\nuv add ruff\nuv run ruff check\nuv lock\nuv sync\n```\n\n### Scripts\n\n`uv` can manage dependencies for single-file scripts with inline metadata:\n\n```bash\necho \'import requests; print(requests.get("https://astral.sh"))\' > example.py\nuv add --script example.py requests\nuv run example.py\n```\n\n### Tools\n\n`uv` can execute and install co妹妹and-line tools provided by Python packages, similar to `pipx`:\n\n```bash\nuvx pycowsay \'hello world!\'\nuv tool install ruff\n```\n\n### Python Versions\n\n`uv` can install and manage Python versions:\n\n```bash\nuv python install 3.10 3.11 3.12\nuv venv --python 3.12.0\nuv python pin 3.11\n```\n\n### The `pip` Interface\n\n`uv` provides a drop-in replacement for co妹妹on `pip`, `pip-tools`, and `virtualenv` co妹妹ands:\n\n```bash\nuv pip compile docs/requirements.in --universal --output-file docs/requirements.txt\nuv venv\nuv pip sync docs/requirements.txt\n```\n\n## Learn More\n\nFor detailed documentation, visit [docs.astral.sh/uv](https://docs.astral.sh/uv).\n\nYou can also explore the [first steps](https://docs.astral.sh/uv/getting-started/first-steps/) or jump straight to the [guides](https://docs.astral.sh/uv/guides/).'}
AI: I have successfully created a markdown file titled `uv_package_management.md` that explains how to use `uv` to manage Python packages. The file is saved in the `/Users/loong_zhou/mcp/test-mcp/mcp_demo/client/downloads/` directory.
You can now access the file at the specified location for reference or further editing. Let me know if you need any additional assistance!
能够瞅到那边挪用了三次 fetch ，一次 list_allowed_directories 战二次 write_file，终极天生了markdown文献到downloads目次下，并前去疑息给尔。
目前念要进修甚么新常识，就能够间接让它去帮助读与网页，而且归纳后天生文档啦。
One More Thing
您或许会好奇正在那个client的运行中，发作了哪些接互，无妨使用如下号令去理解挪用细节：
LOG_LEVEL=debug python client.py归纳
来日诰日咱们进修了怎样使用DeepSeek动作LLM去创立一个MCP客户端，期望对于您有辅佐。后绝尔会分享一点儿故意思的MCP Server，瞅瞅尔一样平常是怎样调整一点儿数据干阐发的。
假设您以为那篇文章对于您有辅佐，欢送面赞珍藏，感谢撑持！
引用链交
[1] mcp-build-rich-context-ai-apps-with-anthropic: https://learn.deeplearning.ai/courses/mcp-build-rich-context-ai-apps-with-anthropic |
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
照妖镜
广告
