MCP Lite

阀门

用于构建服务器的模型上下文协议（MCP）的Python实现，这些服务器向大型语言模型（LLM）应用暴露工具、资源和提示。

项目目的

Valve提供了一个框架，用于创建符合MCP（机器学习通信协议）规范的服务器，并将其集成到大型语言模型（LLM）应用中。该框架处理协议协商、消息路由、传输层抽象（标准输入输出、直接传输、服务器发送事件），并提供了一个主机编排层，能够同时管理多个MCP服务器。通过这个框架，大型语言模型能够通过标准化协议与外部数据源和工具进行交互。

架构概述

• MCPLite（注：这个名称可能是一个特定产品或服务的名称，直接翻译为“轻量级MCPL”可能不够准确，具体翻译需结合上下文或产品/服务的实际含义，但在此提供一个基本的直译版本）用于定义MCP服务器的服务器端框架，采用装饰器方式。提供 @tool， @resource，以及 @prompt 用于注册功能的装饰器。
• 主机编排引擎，用于管理多个MCP客户端，聚合功能，生成系统提示，并实现与大型语言模型（LLM）交互的代理循环。
• 客户MCP客户端实现，用于连接服务器、进行能力协商并发送请求。
• 服务器消息处理器，用于将传入的MCP请求路由到已注册的原始操作（或基本元素），并返回响应。
• 传输层通信抽象包括DirectTransport（进程内通信）、StdioTransport（通过标准输入输出的子进程通信）和SSETransport（HTTP服务器发送事件）。
• 基本元素/原语核心MCP抽象包括MCPTool、MCPResource、MCPResourceTemplate和MCPPrompt，它们用于将Python函数与MCP元数据封装在一起。
• 信息符合JSON-RPC 2.0规范的MCP协议消息（包括请求、响应、通知和错误）的Pydantic模型。
• 注册表ServerRegistry 存储带有可执行代码的服务器端基本组件；ClientRegistry 存储用于能力发现的客户端定义。
• 库存服务器发现与管理系统，用于扫描目录以查找可用的MCP服务器并维护元数据。
• MCPChat基于Chain框架构建的聊天界面，为对话式大型语言模型（LLM）应用增添了MCP（多渠道平台/管理控制协议等，具体含义根据上下文确定）功能。

依赖项

主要依赖项：

• pydantic数据验证和序列化
• requests用于fetch服务器的HTTP客户端
• beautifulsoup4HTML解析
• markdownifyHTML 转 Markdown
• fastapi用于SSE传输的Web框架
• sse-starletteFastAPI 的 SSE 支持
• aiohttp异步HTTP客户端
• rich终端格式化和输出

本地依赖项：

• Chain提供模型（Model）、消息（Message）、消息存储（MessageStore）、提示（Prompt）和聊天（Chat）类的大型语言模型（LLM）框架（似乎是内部依赖项）

API 文档

MCPLite

class MCPLite:
    def __init__(self, transport: Optional[Transport | str] = None)

用于创建MCP服务器的主要类。

关键方法：

def tool(self, func: Callable) -> Callable

装饰器，用于将函数注册为MCP工具。函数必须包含类型注解和文档字符串。

def resource(self, uri: str, mime_type: str = "text/plain", size: int = 1024) -> Callable

装饰器，用于将函数注册为MCP资源或资源模板。使用 {param} 在模板的URI中。

def prompt(self, func: Callable) -> Callable

装饰器，用于将函数注册为MCP提示。函数应返回一个字符串或PromptMessage对象的列表。

def run(self)

启动服务器。行为取决于传输类型。

主持人

class Host:
    def __init__(
        self,
        servers: list[str],
        model: str = "gpt",
        preferred_transport: transport_types = "stdio",
        console: Console = Console()
    )

用于管理多个MCP服务器的编排引擎。

参数：

• servers要连接的服务器名称列表
• model大型语言模型（LLM）的模型标识符
• preferred_transport传输类型偏好（“stdio”，“direct”，“sse”）
• console丰富的控制台输出

关键方法：

def agent_query(self, prompt: str, message_store: MessageStore = MessageStore()) -> str | None

执行一个代理循环，处理MCP工具调用。返回最终答案或None。

客户

class Client:
    def __init__(
        self,
        name: str = "Generic Client",
        transport: str | Transport | StdioClientTransport = "DirectTransport",
        server_function: Optional[Callable] = None
    )

用于连接服务器的MCP客户端。

关键方法：

def initialize(self)

执行MCP握手和能力发现。

def send_request(self, request: MCPRequest) -> MCPResult

向服务器发送请求并返回结果。

def send_notification(self, notification: MCPNotification)

向服务器发送通知（不期待响应）。

MCPChat（注：MCPChat可能是一个特定应用或服务的名称，直接翻译为“MCP聊天”可能不够准确，具体翻译需根据上下文或该名称的实际含义来确定。在此仅提供直译版本。）

class MCPChat(Chat):
    def __init__(
        self,
        servers: list[str],
        model: str = "gpt",
        preferred_transport: transport_types = "stdio",
        **kwargs
    )

具有MCP功能的聊天界面。继承自Chain的Chat类。

参数：

• servers要连接的MCP服务器名称列表
• model模型标识符
• preferred_transport交通方式偏好

附加命令：

• /status显示MCP连接状态
• /list_tools列出可用工具
• /list_resources列出可用资源
• /list_prompts列出可用的提示

运输类别

class StdioClientTransport(Transport):
    def __init__(self, server_command: list[str])

通过标准输入输出进行子进程通信的客户端传输。

class StdioServerTransport(Transport):
    def __init__(self)

用于标准I/O通信的服务器传输。

class DirectTransport(Transport):
    def __init__(self, server_function: Callable)

直接调用服务器函数的中间过程传输。

使用示例

创建MCP服务器

from valve.mcplite.mcplite import MCPLite
from valve.transport import StdioServerTransport

mcp = MCPLite(transport=StdioServerTransport())

@mcp.tool
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@mcp.resource(uri="myapp://status")
def get_status() -> str:
    """Get application status."""
    return "Server is running"

@mcp.prompt
def analysis_prompt(topic: str) -> str:
    """Generate an analysis prompt for a given topic."""
    return f"Please analyze the following topic in detail: {topic}"

if __name__ == "__main__":
    mcp.run()

使用主机来协调多个服务器

from valve.host.Host import Host
from Chain import MessageStore

# Connect to multiple MCP servers
host = Host(
    model="gpt",
    servers=["fetch", "obsidian"],
    preferred_transport="stdio"
)

# Execute a query that may use multiple tools
message_store = MessageStore()
result = host.agent_query(
    "Fetch the content from example.com and save it to my notes",
    message_store
)
print(result)

使用MCP构建聊天应用程序

from valve.mcpchat.mcpchat import MCPChat

# Create chat with MCP capabilities
chat = MCPChat(
    model="gpt",
    servers=["fetch", "obsidian"],
    preferred_transport="direct"
)

# Start interactive chat
chat.chat()

# Or use programmatically
response = chat.query("What's the weather in San Francisco?")