最近很火的MCP是什么？一文帶你搞懂，附大量MCP開源服務(wù)端項目！

網(wǎng)摘文苑 2025-04-27

展開全文

大模型擁有豐富的通用知識，但卻無法訪問專屬內(nèi)容，無法聯(lián)網(wǎng)獲取實時信息。它的核心功能是生成token，無法精細(xì)化操作具體的任務(wù)。

而MCP則提供了一套標(biāo)準(zhǔn)協(xié)議來解決這些問題，換句話說，通過一系列外部工具，可以幫助大模型獲取無法知曉的信息或者難以執(zhí)行的操作。

今天讓我們來了解一下什么是MCP。

什么是MCP

MCP（Model Context Protocol) 是一種開放協(xié)議，它為應(yīng)用程序和大模型之間交互提供了標(biāo)準(zhǔn)化的接口?？梢园袽CP想象成人工智能應(yīng)用程序的USB-C端口，它提供了一種將人工智能模型連接到不同數(shù)據(jù)源和工具的標(biāo)準(zhǔn)化方式。

為什么需要有MCP

MCP 能幫助你在大語言模型之上構(gòu)建代理和復(fù)雜工作流。大語言模型常常需要與數(shù)據(jù)和工具集成，而 MCP 提供：

一長串預(yù)先構(gòu)建的集成，你的大語言模型可以直接接入
在不同的大語言模型提供商和供應(yīng)商之間切換的靈活性
在你的基礎(chǔ)設(shè)施內(nèi)保護(hù)數(shù)據(jù)的最佳實踐方法

從本質(zhì)上講，MCP 遵循客戶端 - 服務(wù)器架構(gòu)，在這種架構(gòu)下，一個主機(jī)應(yīng)用程序可以連接到多個服務(wù)器。

MCP Hosts（MCP主機(jī)）：想要通過 MCP 訪問數(shù)據(jù)的 Claude Desktop、IDE 或 AI 工具等程序
MCP Client（MCP客戶端）：與服務(wù)器保持 1:1 連接的協(xié)議客戶端
MCP Servers（MCP服務(wù)端）：輕量級程序，每個程序都通過標(biāo)準(zhǔn)化的模型上下文協(xié)議公開特定功能
Local Data Sources（本地數(shù)據(jù)源）：MCP 服務(wù)器可以安全訪問的計算機(jī)文件、數(shù)據(jù)庫和服務(wù)
Remote Services（遠(yuǎn)程服務(wù)）：MCP 服務(wù)器可以連接到的 Internet 上可用的外部系統(tǒng)（例如，通過 API）

另外，本地服務(wù)和云端服務(wù)的區(qū)別：

當(dāng) MCP 服務(wù)器與本地安裝的軟件通信時使用本地服務(wù)，例如控制 Chrome 瀏覽器。
當(dāng) MCP 服務(wù)器與遠(yuǎn)程 API 通信時使用網(wǎng)絡(luò)服務(wù)，例如天氣 API。

客戶端集成MCP服務(wù)端的例子

下面介紹一下如何在客戶端集成MCP服務(wù)端（以Python代碼為例），下面以構(gòu)建一個LLM驅(qū)動的聊天機(jī)器人客戶端為例，該客戶端連接到 MCP 服務(wù)器。

使用uv創(chuàng)建一個新的Python項目

# Create project directoryuv init mcp-clientcd mcp-client# Create virtual environmentuv venv# Activate virtual environment# On Windows:.venv\Scripts\activate# On Unix or MacOS:source .venv/bin/activate# Install required packagesuv add mcp anthropic python-dotenv# Remove boilerplate filesrm hello.py# Create our main filetouch client.py

設(shè)置一下API Key

# Create .env file touch .env

然后再.env中添加key

ANTHROPIC_API_KEY=<your key here>

創(chuàng)建一個客戶端，下面是一個基礎(chǔ)框架

import asynciofrom typing import Optionalfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom anthropic import Anthropicfrom dotenv import load_dotenvload_dotenv()  # load environment variables from .envclass MCPClient:    def __init__(self):        # Initialize session and client objects        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()        self.anthropic = Anthropic()    # methods will go here

然后再搞一個服務(wù)端連接管理器：

async def connect_to_server(self, server_script_path: str):    '''Connect to an MCP server    Args:        server_script_path: Path to the server script (.py or .js)    '''    is_python = server_script_path.endswith('.py')    is_js = server_script_path.endswith('.js')    if not (is_python or is_js):        raise ValueError('Server script must be a .py or .js file')    command = 'python' if is_python else 'node'    server_params = StdioServerParameters(        command=command,        args=[server_script_path],        env=None    )    stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))    self.stdio, self.write = stdio_transport    self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))    await self.session.initialize()    # List available tools    response = await self.session.list_tools()    tools = response.tools    print('\nConnected to server with tools:', [tool.name for tool in tools])

查詢的處理邏輯：

async def process_query(self, query: str) -> str:    '''Process a query using Claude and available tools'''    messages = [        {            'role': 'user',            'content': query        }    ]    response = await self.session.list_tools()    available_tools = [{        'name': tool.name,        'description': tool.description,        'input_schema': tool.inputSchema    } for tool in response.tools]    # Initial Claude API call    response = self.anthropic.messages.create(        model='claude-3-5-sonnet-20241022',        max_tokens=1000,        messages=messages,        tools=available_tools    )    # Process response and handle tool calls    final_text = []    assistant_message_content = []    for content in response.content:        if content.type == 'text':            final_text.append(content.text)            assistant_message_content.append(content)        elif content.type == 'tool_use':            tool_name = content.name            tool_args = content.input            # Execute tool call            result = await self.session.call_tool(tool_name, tool_args)            final_text.append(f'[Calling tool {tool_name} with args {tool_args}]')            assistant_message_content.append(content)            messages.append({                'role': 'assistant',                'content': assistant_message_content            })            messages.append({                'role': 'user',                'content': [                    {                        'type': 'tool_result',                        'tool_use_id': content.id,                        'content': result.content                    }                ]            })            # Get next response from Claude            response = self.anthropic.messages.create(                model='claude-3-5-sonnet-20241022',                max_tokens=1000,                messages=messages,                tools=available_tools            )            final_text.append(response.content[0].text)    return '\n'.join(final_text)

互動聊天界面：

async def chat_loop(self):    '''Run an interactive chat loop'''    print('\nMCP Client Started!')    print('Type your queries or 'quit' to exit.')    while True:        try:            query = input('\nQuery: ').strip()            if query.lower() == 'quit':                break            response = await self.process_query(query)            print('\n' + response)        except Exception as e:            print(f'\nError: {str(e)}')async def cleanup(self):    '''Clean up resources'''    await self.exit_stack.aclose()

定義main函數(shù)：

async def main():    if len(sys.argv) < 2:        print('Usage: python client.py <path_to_server_script>')        sys.exit(1)    client = MCPClient()    try:        await client.connect_to_server(sys.argv[1])        await client.chat_loop()    finally:        await client.cleanup()if __name__ == '__main__':    import sys    asyncio.run(main())