MCPLite

valve

用于构建 MCP 服务的 Python 实现，这些服务可向大模型应用开放工具、资源和提示词。

项目用途

Valve 提供了一个框架，用于创建符合 MCP 规范的服务并将其集成到大模型应用中。它负责处理协议协商、消息路由、传输层抽象（stdio、direct、SSE），并提供可同时管理多个 MCP 服务的主机编排层。该框架使大模型能够通过标准化协议与外部数据源和工具交互。

架构概览

• MCPLite：用于通过装饰器定义 MCP 服务的服务端框架。提供 @tool、@resource 和 @prompt 装饰器，用于注册功能。
• Host：编排引擎，可管理多个 MCP 客户端、汇总功能、生成系统提示词，并为大模型交互实现 Agent 循环。
• Client：MCP 客户端实现，可连接服务、执行功能协商并发送请求。
• Server：消息处理器，可将传入的 MCP 请求路由到已注册的原语并返回响应。
• Transport Layer：通信抽象，包括 DirectTransport（进程内）、StdioTransport（通过 stdio 的子进程）和 SSETransport（HTTP SSE）。
• Primitives：核心 MCP 抽象，包括 MCPTool、MCPResource、MCPResourceTemplate 和 MCPPrompt，它们会为 Python 函数封装 MCP 元数据。
• Messages：用于 MCP 协议消息的 Pydantic 模型，包括符合 JSON-RPC 2.0 的请求、响应、通知和错误。
• Registry：ServerRegistry 保存含可执行代码的服务端原语；ClientRegistry 保存用于功能发现的客户端定义。
• Inventory：服务发现和管理系统，可扫描目录以查找可用的 MCP 服务并维护元数据。
• MCPChat：基于 Chain 框架构建的聊天界面，可为对话式大模型应用添加 MCP 功能。

依赖项

主要依赖项：

• pydantic：数据验证和序列化
• requests：用于 fetch 服务的 HTTP 客户端
• beautifulsoup4：HTML 解析
• markdownify：将 HTML 转换为 Markdown
• fastapi：用于 SSE 传输的 Web 框架
• sse-starlette：为 FastAPI 提供 SSE 支持
• aiohttp：异步 HTTP 客户端
• rich：终端格式化和输出

本地依赖项：

• Chain：提供 Model、Message、MessageStore、Prompt 和 Chat 类的大模型框架（似乎是内部依赖项）

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 资源或资源模板的装饰器。对于模板，请在 URI 中使用 {param}。

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

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

def run(self)

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

Host

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

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

参数：

• servers：要连接的服务名称列表
• model：大模型的模型标识符
• preferred_transport：首选传输类型（"stdio"、"direct"、"sse"）
• console：用于输出的 Rich 控制台

主要方法：

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

执行处理 MCP 工具调用的 Agent 循环。返回最终答案或 None。

Client

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

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

通过 stdio 进行子进程通信的客户端传输。

class StdioServerTransport(Transport):
    def __init__(self)

用于 stdio 通信的服务端传输。

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

使用 Host 协调多个服务

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?")