在人工智能快速發(fā)展的今天,大型語(yǔ)言模型(LLM)的應(yīng)用場(chǎng)景不斷拓展。為了讓LLM更好地整合數(shù)據(jù)和工具,MCP(Model Context Protocol)應(yīng)運(yùn)而生。它就像AI領(lǐng)域的USB-C接口,為AI應(yīng)用提供了標(biāo)準(zhǔn)化的連接方式。今天,我們將為大家?guī)?lái)一篇適合國(guó)內(nèi)零基礎(chǔ)小白學(xué)習(xí)的MCP客戶端開(kāi)發(fā)教程,幫助大家輕松上手MCP客戶端開(kāi)發(fā)。
在開(kāi)始之前,你需要確保系統(tǒng)滿足以下要求:
uv
。
首先,我們需要?jiǎng)?chuàng)建一個(gè)新的Python項(xiàng)目并安裝uv
:
pip install uv
你需要從Anthropic控制臺(tái)獲取一個(gè)API密鑰。
.env
文件來(lái)存儲(chǔ)它:
touch .env
.env
文件中:
ANTHROPIC_API_KEY=your_api_key_here
.env
添加到你的.gitignore
文件中,以避免將其提交到Git倉(cāng)庫(kù):
.env
首先,我們?cè)O(shè)置導(dǎo)入并創(chuàng)建基本的客戶端類:
import os
import asyncio
from typing import List, Optional
from dotenv import load_dotenv
from anthropic import ClaudeApi
from fastmcp import FastMCPClient
load_dotenv()
class MCPChatClient:
def __init__(self):
self.claude_api = ClaudeApi(api_key=os.getenv("ANTHROPIC_API_KEY"))
self.mcp_client = FastMCPClient()
self.session = None
接下來(lái),我們實(shí)現(xiàn)連接到MCP服務(wù)器的方法:
async def connect_to_server(self, server_path: str):
"""連接到MCP服務(wù)器"""
await self.mcp_client.connect(server_path)
self.session = await self.mcp_client.initialize_session()
tools = await self.mcp_client.list_tools(self.session)
print(f"Available tools: {tools}")
現(xiàn)在,我們添加處理查詢和處理工具調(diào)用的核心功能:
async def process_query(self, query: str) -> str:
"""處理查詢并調(diào)用工具"""
# 獲取可用工具
tools = await self.mcp_client.list_tools(self.session)
# 發(fā)送查詢到Claude并獲取響應(yīng)
response = await self.claude_api.send_query(query, tools)
# 處理工具調(diào)用
if response.requires_tool_call:
tool_call_result = await self.mcp_client.call_tool(
self.session,
response.tool_to_call,
response.tool_input
)
response = await self.claude_api.send_tool_result(
response.conversation_id,
tool_call_result
)
return response.content
現(xiàn)在,我們添加聊天循環(huán)和清理功能:
async def chat(self):
"""啟動(dòng)交互式聊天界面"""
print("Starting chat with Claude...")
print("Type 'exit' to end the chat.")
while True:
query = input("You: ")
if query.lower() == 'exit':
break
response = await self.process_query(query)
print(f"Claude: {response}")
await self.mcp_client.close()
最后,我們添加主執(zhí)行邏輯:
async def main():
client = MCPChatClient()
server_path = input("Enter the path to your MCP server: ")
await client.connect_to_server(server_path)
await client.chat()
if __name__ == "__main__":
asyncio.run(main())
MCPChatClient
類初始化會(huì)話管理和API客戶端。AsyncExitStack
進(jìn)行正確的資源管理。process_query()
以處理特定類型的工具。要運(yùn)行你的客戶端與任何MCP服務(wù)器:
客戶端將:
當(dāng)你提交一個(gè)查詢時(shí):
AsyncExitStack
進(jìn)行正確的清理。.env
中。如果你看到:
FileNotFoundError
:檢查你的服務(wù)器路徑。Connection refused
:確保服務(wù)器正在運(yùn)行且路徑正確。Tool execution failed
:驗(yàn)證工具所需的環(huán)境變量是否已設(shè)置。Timeout error
:考慮在客戶端配置中增加超時(shí)時(shí)間。希望這篇教程能幫助你更好地理解和掌握MCP客戶端開(kāi)發(fā)。在編程獅(W3Cschool.cn)平臺(tái),你可以找到更多關(guān)于MCP開(kāi)發(fā)的實(shí)例和教程,幫助你進(jìn)一步提升開(kāi)發(fā)技能。
更多建議: