最近很火的MCP是什么？一文带你搞懂，附大量MCP开源服务端项目！

网摘文苑 2025-04-27

展开全文

大模型拥有丰富的通用知识，但却无法访问专属内容，无法联网获取实时信息。它的核心功能是生成token，无法精细化操作具体的任务。

而MCP则提供了一套标准协议来解决这些问题，换句话说，通过一系列外部工具，可以帮助大模型获取无法知晓的信息或者难以执行的操作。

今天让我们来了解一下什么是MCP。

什么是MCP

MCP（Model Context Protocol) 是一种开放协议，它为应用程序和大模型之间交互提供了标准化的接口。可以把MCP想象成人工智能应用程序的USB-C端口，它提供了一种将人工智能模型连接到不同数据源和工具的标准化方式。

为什么需要有MCP

MCP 能帮助你在大语言模型之上构建代理和复杂工作流。大语言模型常常需要与数据和工具集成，而 MCP 提供：

一长串预先构建的集成，你的大语言模型可以直接接入
在不同的大语言模型提供商和供应商之间切换的灵活性
在你的基础设施内保护数据的最佳实践方法

从本质上讲，MCP 遵循客户端 - 服务器架构，在这种架构下，一个主机应用程序可以连接到多个服务器。

MCP Hosts（MCP主机）：想要通过 MCP 访问数据的 Claude Desktop、IDE 或 AI 工具等程序
MCP Client（MCP客户端）：与服务器保持 1:1 连接的协议客户端
MCP Servers（MCP服务端）：轻量级程序，每个程序都通过标准化的模型上下文协议公开特定功能
Local Data Sources（本地数据源）：MCP 服务器可以安全访问的计算机文件、数据库和服务
Remote Services（远程服务）：MCP 服务器可以连接到的 Internet 上可用的外部系统（例如，通过 API）

另外，本地服务和云端服务的区别：

当 MCP 服务器与本地安装的软件通信时使用本地服务，例如控制 Chrome 浏览器。
当 MCP 服务器与远程 API 通信时使用网络服务，例如天气 API。

客户端集成MCP服务端的例子

下面介绍一下如何在客户端集成MCP服务端（以Python代码为例），下面以构建一个LLM驱动的聊天机器人客户端为例，该客户端连接到 MCP 服务器。

使用uv创建一个新的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

设置一下API Key

# Create .env file touch .env

然后再.env中添加key

ANTHROPIC_API_KEY=<your key here>

创建一个客户端，下面是一个基础框架

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

然后再搞一个服务端连接管理器：

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函数：

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())