优秀的编程知识分享平台

网站首页 > 技术文章 正文

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

nanyue 2025-03-25 16:01:23 技术文章 17 ℃

大家好,我是Jack Bytes,一个专注于将人工智能应用于日常生活的程序猿,平时主要分享AI、NAS、开源项目等。

如果你对本文内容感兴趣,欢迎点赞、收藏、转发,你的支持就是我最大的动力!

大模型拥有丰富的通用知识,但却无法访问专属内容,无法联网获取实时信息。它的核心功能是生成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 directory
uv init mcp-client
cd mcp-client

# Create virtual environment
uv venv

# Activate virtual environment
# On Windows:
.venv\Scripts\activate
# On Unix or MacOS:
source .venv/bin/activate

# Install required packages
uv add mcp anthropic python-dotenv

# Remove boilerplate files
rm hello.py

# Create our main file
touch client.py

设置一下API Key

# Create .env file 
touch .env

然后再.env中添加key

ANTHROPIC_API_KEY=

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

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from .env

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

然后执行下面命令运行客户端:

uv run client.py path/to/server.py # python server

MCP服务器精选

在Github上,开源项目punkpeye/awesome-mcp-servers收集了大量的MCP开源项目,如下所示:

大模型学习资料

我整理了一份大模型学习相关的资料,不同于网上分享的大杂烩资源,我将资源进行了整理,目录结构一目了然,对新手更加友好,后续也会持续更新!

放在公众号了,感兴趣的兄弟可以自行取用!获取关键词为:「大模型学习资料

我是Jack Bytes

一个专注于将人工智能应用于日常生活的半吊子程序猿!

平时主要分享AI、NAS、Docker、搞机技巧、开源项目等技术,喜欢的话请关注吧!

最近发表
标签列表