MCP 服务器实现：FastMCP 工具注册与 REST 桥接

学习目标

• 理解 Langflow 独立 MCP Server 的 FastMCP 实现
• 掌握组件注册表的懒加载和 contextvars 并发隔离
• 了解 Flow Builder 工具集的原子操作
• 理解 MCP 工具调用与 TelemetryService 的集成

前置知识

本章涉及 MCP 协议的通用原理，建议先阅读：

• MCP 协议与生态集成

下文假设你已理解 MCP 的基本协议和工具调用流程，直接聚焦 Langflow 的具体实现。

---

项目实践

1. FastMCP Server 架构

独立 MCP Server 基于 mcp.server.fastmcp.FastMCP 构建：

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("langflow")

@mcp.tool()
async def list_components(ctx: Context, query: str = None) -> list[dict]:
    """搜索可用的 Langflow 组件。"""
    registry = await get_registry(ctx)
    return search_registry(registry, query)

工具分组：

组	工具	功能
Auth	authenticate	验证 API Key 和连接状态
Component	list_components, describe_component	组件搜索和描述
Flow	add_component, remove_component, configure_component	Flow 构建的原子操作
Connection	add_connection, remove_connection	组件间连线管理
Execution	run_flow, flow_to_spec_summary	Flow 执行与导出
Batch	批量操作工具	一次修改多个组件或连线

2. 组件注册表懒加载

MCP Server 首次访问时从 Langflow API 拉取完整组件注册表：

多 Agent 并发隔离（SSE 模式）：

# contextvars 实现 per-session 隔离
_client_var: contextvars.ContextVar[LangflowClient | None] = contextvars.ContextVar("_client", default=None)
_registry_var: contextvars.ContextVar[dict | None] = contextvars.ContextVar("_registry", default=None)

# stdio 模式（单 Agent）使用模块级共享
_shared_client: LangflowClient | None = None
_shared_registry: dict | None = None

获取注册表的辅助函数：

async def get_registry(ctx: Context) -> dict:
    # 优先从 contextvar 读取（SSE 模式）
    registry = _registry_var.get()
    if registry is not None:
        return registry
    # 回退到模块级共享（stdio 模式）
    return _shared_registry

3. Flow Builder 工具集

Flow Builder 工具提供对 Flow JSON 的原子操作：

# src/lfx/src/lfx/graph/flow_builder/ 中的函数

# 添加组件到 Flow
def add_component(flow_spec: dict, component_type: str, position: dict) -> dict: ...

# 在两个组件间建立连线
def add_connection(flow_spec: dict, source_id: str, target_id: str, handle: dict) -> dict: ...

# 修改组件配置
def configure_component(flow_spec: dict, component_id: str, field: str, value: Any) -> dict: ...

# 导出 Flow 为可读摘要
def flow_to_spec_summary(flow_spec: dict) -> str: ...

# 将 Flow 导出为 ASCII 图
def flow_graph_repr(flow_spec: dict) -> str: ...

零缓存策略的执行流程：

每次操作都重新获取最新 Flow，确保多 Agent 并发修改时的强一致性。

4. 遥测集成

每次 MCP 工具调用自动上报到 TelemetryService：

from lfx.services.telemetry import MCPToolPayload, TelemetryService

@mcp.tool()
async def list_components(ctx: Context, query: str = None):
    # ... 业务逻辑
    if _telemetry:
        _telemetry.track_mcp_tool_call(MCPToolPayload(
            tool_name="list_components",
            query=query,
            result_count=len(results),
        ))

上报内容：

• 工具名、参数（脱敏后）、结果数量
• 调用时间戳、Session ID
• 错误信息（如果调用失败）

---

问题与规避

陷阱	现象	对策
注册表未初始化	首次调用报错 “Registry not loaded”	确保 Langflow API 已启动且 MCP Server 的 --url 正确
Flow JSON 并发冲突	两个 Agent 同时修改同一 Flow，后执行的覆盖先执行的	零缓存策略降低冲突概率，但业务层仍需注意避免同时修改
工具描述过短	Agent 无法理解工具用途	在 @mcp.tool() 装饰器中编写详细的 docstring
遥测数据泄露敏感信息	工具参数中包含 API Key 或用户数据	使用 redact 工具过滤敏感字段后再上报

---

设计取舍

FastMCP vs 手写 MCP Server：

维度	FastMCP	手写 MCP Server
开发成本	低，装饰器定义工具	高，需手动实现 JSON-RPC 处理
工具类型推导	自动从函数签名推导	手动编写 JSON Schema
灵活性	受限于 FastMCP 的抽象	完全控制协议细节
版本兼容	依赖 FastMCP SDK 更新	自行处理协议版本变更

Langflow 选择 FastMCP 的原因是工具集相对固定（Flow 构建原子操作），且需要快速迭代。FastMCP 的装饰器风格使工具定义和实现集中在同一个函数中，降低了维护成本。

---

参考来源

• 源码：src/lfx/src/lfx/mcp/server.py
• 源码：src/lfx/src/lfx/mcp/flow_builder_tools.py
• 源码：src/lfx/src/lfx/mcp/registry.py
• 源码：src/lfx/src/lfx/services/telemetry.py