MCP网关：发现+生成LangGraph工作流
演示存储库：该项目演示了用于API端点发现和MCP工具生成的LangGraph工作流编排。它仅用于教育和参考目的。
一个强大的两阶段工作流系统，可自动从规范或文档中发现API端点，并生成符合JSON Schema Draft-07的MCP（模型上下文协议）工具定义。
目录
______________________________________________________________________
概述
此项目实现了两个可组合的LangGraph工作流：
- DiscoveryGraph -从以下位置智能地发现API终结点：
- OpenAPI/Swagger规范（JSON/YAML） - 文档网站（通过网络爬虫） - 自由格式文本文件（通过LLM提取）
- 生成图 -生成生产就绪的MCP工具定义，包括：
- 符合JSON模式Draft-07的参数模式 - 全面的验证和错误处理 - 从端点到工具的自动化工作流程
该系统利用 LangGraph 用于工作流编排和 LangChain LLM操作。这两个图都在没有中断的情况下运行到完成，端点选择在编排层中处理。
______________________________________________________________________
主要特点
- 多源发现：处理OpenAPI规范、web文档和非结构化文本
- 混合萃取：将正则表达式模式匹配与LLM驱动的提取相结合
- 智能网络爬行：具有域边界和速率限制的站点地图感知爬行
- 智能重复数据删除：对发现的端点进行标准化和重复数据消除
- 架构验证：JSON模式草稿-07合规性检查
- 状态持久性：用于工作流恢复的基于SQLite的检查点
- 标准化命名：一致
VENDOR__RESOURCE__VERB工具命名约定
______________________________________________________________________
建筑
双图工作流系统
+-------------------------------------------------------------------+
| INPUT SOURCES |
| * OpenAPI/Swagger Specs (JSON/YAML) |
| * Documentation URLs |
| * Free-form text files |
+-----------------------------+-------------------------------------+
|
v
+-------------------------------------------------------------------+
| DISCOVERY GRAPH |
| Discovers and catalogs API endpoints |
+-----------------------------+-------------------------------------+
|
v
[Orchestration Layer]
(Endpoint Selection)
|
v
+-------------------------------------------------------------------+
| GENERATION GRAPH |
| Generates MCP tool definitions |
+-----------------------------+-------------------------------------+
|
v
>> MCP Tools (JSON) 正则表达式>LLM）
- 描述的存在
- 参数完整性
- 创建分层目录结构
- 提供摘要统计信息
- 完成发现图工作流
______________________________________________________________________
### 生成图形流
生成图由6个节点组成，这些节点将选定的端点转换为经过验证的MCP工具：
plan_work -> schema_synthesis -> compose_tool -> validate -> aggregate_tools -> finalize -> END
#### 节点分解
|节点|目的|输入|输出|
|------|---------|-------|--------|
| **计划工作** |为每个选定端点创建工作项|选定端点ID |工作项队列|
| **方案综合** |使用LLM生成JSON模式参数|端点详细信息|参数模式|
| **compose_tool** |使用命名|模式+端点|MCP工具对象构建完整的MCP工具结构|
| **验证** |根据JSON Schema Draft-07验证模式|工具定义|仅验证工具|
| **聚合工具** |对所有已验证的工具进行排序和聚合|单个工具|已排序的工具集合|
| **完成** |标记完成并提供摘要|已验证的工具|最终工具列表|
#### 生成节点详细信息
**1.计划_工作_节点** (`generation/nodes.py:34`)
- 从所选端点ID创建工作项
- 筛选端点以仅包括选定的端点
- 初始化状态跟踪（“待定”、“schema_generated”等）
- 设置错误收集结构
**2.模式_合成_节点** (`generation/nodes.py:60`)
- 为每个端点使用带结构化提示的LLM
- 生成包含4个组的嵌套参数架构：
- `header` -HTTP标头
- `path` -路径参数
- `query` -查询字符串参数
- `body` -请求正文
- 强制JSON模式功能：
- 类型约束和格式
- 受限字段的枚举值
- 最小/最大长度和值约束
- 所需字段规范
- 使用数组表示法的null类型
- 使用自定义元数据增强架构：
- `visible`：UI可见字段数组
- `required`：必填字段名
- `additionalProperties`：布尔值用于严格验证
- 优雅地处理错误，标记失败的项目
**3.组合工具节点** (`generation/nodes.py:153`)
- 从服务器URL中提取供应商（域或“未知”）
- 从路径中提取资源（版本后的第一段）
- 根据方法+路径分析确定动词
- 编写工具名称： `VENDOR__RESOURCE__VERB` （大写）
- 根据描述生成人类可读的显示名称
- 构建完整的MCP工具结构：
- 名称、显示名称、描述
- 标签：\[供应商，资源，版本，方法\]
- 协议：“休息”
- 协议数据：方法、路径、服务器url
- 参数：嵌套架构
- 元数据：来源和置信度得分
**4.validate_node** (`generation/nodes.py:225`)
- 创建架构的干净副本（删除自定义 `visible` 现场）
- 使用JSON Schema Draft-07进行验证 `jsonschema` 图书馆
- 用途 `Draft7Validator.check_schema()` 用于验证
- 过滤掉无效工具
- 将验证错误记录到错误数组中
- 使用有效工具的自定义字段保留原始架构
**5.聚合工具节点** (`generation/nodes.py:264`)
- 收集所有经过验证的工具
- 按工具名称的字母顺序排序
- 为输出提供一致的排序
**6.finalize_node** (`generation/nodes.py:284`)
- 将生成标记为“已完成”
- 提供最终工具计数
- 报告遇到的任何错误
- 返回完整的输出工具列表
______________________________________________________________________
## 安装
此项目使用 **紫外线** 使用Python 3.12+进行Python包管理。
### 先决条件
- Python 3.12或更高版本
- [紫外线](https://github.com/astral-sh/uv) 包管理器
### 设置
1. **克隆存储库**:
git clone cd mcp-gateway-graph-demo
1. **安装依赖项**:
uv sync
1. **配置环境变量**:
创建一个 `.env` 项目根目录中的文件（使用 `.env.example` 作为模板）：
# Required for LLM operations AZURE_OPENAI_API_KEY=your_api_key_here AZURE_OPENAI_API_VERSION=2024-02-15 AZURE_OPENAI_ENDPOINT=https://your-endpoint.openai.azure.com/
# Optional: LangSmith tracing LANGSMITH_API_KEY=your_langsmith_key LANGSMITH_TRACING=true LANGSMITH_PROJECT=mcp-gateway
______________________________________________________________________
## 用法
### 命令行接口
主CLI通过Click提供了几个命令：
Show help
uv run python main.py --help
Run with OpenAPI spec file
uv run python main.py --files api_spec.json --output tools.json
Run with multiple files
uv run python main.py --files spec1.json --files spec2.yaml
Run with documentation URL
uv run python main.py --url https://api.example.com/docs --output tools.json
Run example workflow with mock data
uv run python main.py --example
Auto-approve mode (non-interactive)
uv run python main.py --files api_spec.json --auto-approve --output tools.json
### CLI选项
|选项|描述|
|--------|-------------|
| `-f, --files PATH` |API规范文件的路径（允许多个）|
| `-u, --url TEXT` |要爬网API文档的根URL|
| `-o, --output PATH` |输出文件路径（默认： `mcp_tools.json`) |
| `--auto-approve` |自动选择所有发现的端点（非交互式）|
| `--example` |使用模拟数据运行示例工作流|
### 程序化使用
您还可以直接导入和使用这些模块：
from utils import build_full_workflow
Build both graphs with checkpointing
discovery_graph, generation_graph = build_full_workflow()
Run discovery (runs to completion)
config = {"configurable": {"thread_id": "unique-session-id"}} input_data = { "input": {"files": ["/path/to/spec.json"]}, "discovery": {} }
for event in discovery_graph.stream(input_data, config): print(event)
Get discovered endpoints from final state
final_state = discovery_graph.get_state(config) endpoints = final_state.values.get("discovery", {}).get("endpoints_normalized", [])
Endpoint selection happens in orchestration layer (not in graph)
Then pass selected endpoints to Generation Graph
______________________________________________________________________
## 项目结构
mcp-gateway-graph-demo/ ├── config.py # Configuration, LLM setup, constants ├── main.py # CLI entry point (using Click) ├── .env # Environment variables (not in git) ├── .env.example # Environment template ├── models/ │ ├── __init__.py │ └── schemas.py # State schemas, Pydantic models ├── discovery/ │ ├── __init__.py │ ├── nodes.py # 6 discovery node functions │ ├── helpers.py # OpenAPI parsing, LLM extraction, web crawling │ ├── graph.py # build_discovery_graph() │ └── runners.py # Runner utilities ├── generation/ │ ├── __init__.py │ ├── nodes.py # 6 generation node functions │ ├── helpers.py # Schema enhancement, vendor extraction, validation │ ├── graph.py # build_generation_graph() │ └── runners.py # Runner utilities ├── utils/ │ ├── __init__.py │ ├── workflow.py # build_full_workflow(), checkpointing │ └── tools.py # Utility functions ├── tests/ │ ├── __init__.py │ ├── test_data.py # Mock OpenAPI spec generation │ └── test_workflow.py # Full workflow test └── graph-demo.ipynb # Original notebook (reference only)
______________________________________________________________________
## 配置
### 环境变量
|变量|必填|描述|
|----------|----------|-------------|
| `AZURE_OPENAI_API_KEY` |是|用于LLM操作的Azure OpenAI API密钥|
| `AZURE_OPENAI_API_VERSION` |是| API版本（例如“2024-02-15”）|
| `AZURE_OPENAI_ENDPOINT` |是| Azure OpenAI端点URL|
| `LANGSMITH_API_KEY` |否|用于跟踪的LangSmith API密钥|
| `LANGSMITH_TRACING` |否|启用跟踪（“true”/“false”）|
| `LANGSMITH_PROJECT` |否|LangSmith项目名称（默认：“mcp-gateway”）|
### 常量（config.py）
- `DEFAULT_MODEL = "gpt-4.1"` -LLM模型用于提取和生成
- `MAX_CRAWL_PAGES = 20` -每个域可抓取的最大页面数
- `CRAWL_DELAY = 0.5` -爬网请求之间的秒数
- `LLM_PAGE_SAMPLE_LENGTH = 3000` -从长页面中采样的字符
______________________________________________________________________
## 输出格式
生成的MCP工具遵循标准化的JSON结构：
{ "name": "EXAMPLE__FLIGHTS__SEARCH", "display_name": "Search for available flights", "description": "Search for flights based on origin, destination, and dates", "tags": ["example", "flights", "v1", "post"], "visibility": "public", "active": true, "protocol": "rest", "protocol_data": { "method": "POST", "path": "/api/v1/flights/search", "server_url": "https://api.example.com" }, "parameters": { "header": { "type": "object", "properties": { "Authorization": { "type": "string", "description": "Bearer token for authentication" } }, "required": ["Authorization"], "visible": ["Authorization"] }, "path": { "type": "object", "properties": {}, "required": [], "visible": [] }, "query": { "type": "object", "properties": { "limit": { "type": "integer", "description": "Maximum results to return", "minimum": 1, "maximum": 100, "default": 10 } }, "required": [], "visible": ["limit"] }, "body": { "type": "object", "properties": { "origin": { "type": "string", "description": "Origin airport code", "pattern": "^[A-Z]{3}$" }, "destination": { "type": "string", "description": "Destination airport code", "pattern": "^[A-Z]{3}$" }, "date": { "type": "string", "format": "date", "description": "Departure date" } }, "required": ["origin", "destination", "date"], "visible": ["origin", "destination", "date"] } }, "metadata": { "source": "openapi", "confidence": 0.95 } }
### 工具命名约定
工具遵循以下模式： **`VENDOR__RESOURCE__VERB`** （大写）
示例：
- `STRIPE__PAYMENTS__CREATE`
- `GITHUB__REPOSITORIES__LIST`
- `EXAMPLE__BOOKINGS__DELETE`
### 字段描述
|字段|类型|描述|
|-------|------|-------------|
| `name` |string |唯一工具标识符（VENDOR\_\_RESOURCE\_\_VERB）|
| `display_name` |string | UI显示的人类可读名称|
| `description` |string |工具功能的清晰描述|
| `tags` |array |\[供应商、资源、版本、方法\]|
| `visibility` |string |始终为“public”|
| `active` |boolean |始终为真|
| `protocol` |string |总是“休息”|
| `protocol_data` |object |包含方法、路径和server_url|
| `parameters` |object |头/路径/查询/正文的嵌套架构|
| `metadata` |object |源类型和置信度得分|
______________________________________________________________________
## 发展
### 运行测试
Run full workflow test
uv run python tests/test_workflow.py
Run with example data
uv run python main.py --example
### 代码质量
该项目使用 **拉夫** 用于代码格式化、导入排序和linting。
Format code
uv run ruff format .
Sort imports and fix issues
uv run ruff check --fix .
Lint code (check only)
uv run ruff check .
### 重要实施注意事项
1. **Pydantic版本兼容性**:
- 使用 `from pydantic import BaseModel, Field` （Pydantic v2）
- 请勿使用已弃用的 `langchain_core.pydantic_v1` 导入
1. **Azure OpenAI**:
- 此项目使用Azure OpenAI，而不是标准OpenAI
- 配置所有必需的Azure环境变量
1. **自定义架构字段**:
- 这 `visible` 字段是UI渲染的自定义扩展名
- 在JSON模式验证过程中会自动删除它
1. **检查点**:
- 默认情况下使用内存中的SQLite
- 可以更改为基于文件的 `utils/workflow.py`
1. **类型提示**:
- 所有函数都包含全面的类型提示
- 使用Python `typing` 整个模块
______________________________________________________________________
## 依赖项
关键包（见 `pyproject.toml` 完整列表）：
### 工作流和LLM
- `langgraph>=1.0.1` -工作流编排
- `langchain>=1.0.2` -LLM运营
- `langchain-openai>=1.0.1` -Azure OpenAI集成
- `langgraph-checkpoint-sqlite>=3.0.0` -状态持久性
### CLI和实用程序
- `click>=8.1.0` -现代CLI框架
- `python-dotenv` -环境变量管理
### 数据处理
- `beautifulsoup4>=4.14.2` -用于web发现的HTML/XML解析
- `jsonschema>=4.25.1` -架构验证
- `pyyaml>=6.0.3` -YAML解析
- `requests>=2.32.5` -HTTP请求
### 开发工具
- `ruff>=0.14.3` -代码格式化、导入排序和linting
______________________________________________________________________
## 学习资源
- [LangGraph文档](https://langchain-ai.github.io/langgraph/)
- [LangChain文档](https://python.langchain.com/)
- [JSON模式规范](https://json-schema.org/)
- [OpenAPI规范](https://swagger.io/specification/)
______________________________________________________________________
## 许可证
这是一个用于教育和参考目的的演示项目。请根据您的具体使用情况进行调整。
______________________________________________________________________
## 贡献
这是一个演示存储库。如有疑问或建议，请打开一个问题。
______________________________________________________________________
## 免责声明
**此存储库仅用于演示目的。** 它展示了使用LangGraph的工作流编排模式，应该针对生产用例进行调整和强化。该代码演示了架构概念，可能需要额外的错误处理、安全考虑和实际应用程序的性能优化。
______________________________________________________________________