内容
# OpenLCA MCP 服务器，用于 AI 代理自动化
模型上下文协议（MCP）服务器，公开 openLCA 功能，作为 AI 代理自动化的生命周期评估（LCA）工作流工具。
## 概述
MCP 服务器使 AI 代理（如 n8n 工作流中的代理）能够以编程方式与 openLCA 交互。工具按照 ISO-14040/14044 LCA 方法论的 4 个阶段组织：
1. **目标和范围定义** - 搜索材料、工艺和影响方法
2. **生命周期库存（LCI）** - 创建流、工艺和产品系统
3. **生命周期影响评估（LCIA）** - 计算环境影响
4. **解释** - 分析贡献、不确定性和导出结果
## 功能
- ✅ **24 个专门的 LCA 工具**，适用于 AI 代理，涵盖完整的 openlca-ipc v0.4 接口
- ✅ **基于 FastMCP 构建** — stdio + 可流式 HTTP，类型化的工具模式
- ✅ **引导式提示 + 资源** — LCA 演练和可拉取数据
(`openlca://connections`, `openlca://impact-methods`, `openlca://product-systems`)
- ✅ **多实例** — 命名 [连接配置文件](#connections--multi-tenancy)；一个服务器
可以针对多个 openLCA 实例，每个调用选择一个
- ✅ **身份验证 / 多租户** — 可选的 API 密钥，将调用者映射到允许的实例
- ✅ **按阶段组织**，遵循 ISO LCA 标准
- ✅ **结构化代理响应** — 紧凑的 `ResultSummary`、可重现性上下文和
可恢复的错误信封 (`error_code`, `recoverable`, `suggested_next_actions`)
- ✅ **只读安全模式** (`OPENLCA_READ_ONLY=true`)，保护数据库免受写入
- ✅ **稳定的结果句柄**，带有 `result_id` 注册表，用于后续分析
- ✅ **兼容 n8n / ChatGPT / Claude**，用于工作流自动化
基于 **openlca-ipc ≥ 0.4.0** 及其代理层。
## 先决条件
在运行 MCP 服务器之前：
1. **openLCA 桌面版**
- 安装 2.x 版本并运行
- 加载数据库
- 启动 IPC 服务器（工具 → 开发者工具 → IPC 服务器）
2. **Python 环境**
- Python 3.11 或更高版本
- 安装 openlca-ipc ≥ 0.4.0
3. **用于 n8n 集成**
- 运行 n8n 实例
- 在 n8n 中启用 MCP 集成
## 安装
### 1. 安装依赖项
```bash
cd mcp-server
pip install -r requirements.txt
```
这将安装：
- MCP SDK
- openlca-ipc 库（从父目录）
- Pydantic，用于验证
- python-dotenv，用于配置
### 2. 配置环境
```bash
cp .env.example .env
```
编辑 `.env` 文件：
```bash
# OpenLCA IPC 服务器配置
OPENLCA_PORT=8080 # 匹配您的 IPC 服务器端口
OPENLCA_HOST=localhost
# 安全模式：当为真时，所有数据库写入都被阻止（WRITE_BLOCKED）。
# 读取、搜索和计算仍然有效。默认：false。
OPENLCA_READ_ONLY=false
# 日志配置
LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR
```
### 3. 测试服务器
```bash
python -m src # 或：openlca-mcp (已安装的控制台脚本)
```
您应该看到 FastMCP 横幅、加载的连接配置文件和
到 openLCA 的连接尝试。（`python -m src` — 而不是 `python -m src.app`，后者会重复导入。）
## 连接和多租户
默认情况下，服务器与 **一个** openLCA 实例通信 — `default` 配置文件，基于
`OPENLCA_HOST` / `OPENLCA_PORT` / `OPENLCA_READ_ONLY`（您的本地桌面 openLCA，因此
您创建的实体会出现在 openLCA 界面中以供验证）。
要针对 **多个** 实例，使用一个服务器定义命名的 **连接配置文件** 并
传递 `connection: "<id>"` 到任意工具（省略它则使用默认配置）：
```bash
# 内联 JSON …
OPENLCA_CONNECTIONS='[{"id":"remote","host":"100.107.240.54","port":8080,"kind":"remote","read_only":true}]'
# … 或文件（参见 config/connections.example.json）
OPENLCA_CONNECTIONS_FILE=config/connections.json
```
**身份验证** 默认关闭（开放模式：匿名调用者，仅限 `default` 配置文件）。配置
API 密钥以要求承载令牌并控制每个租户可以使用的配置文件（参见
`config/api_keys.example.json`）：
```bash
OPENLCA_API_KEYS='{"sk_live_xxx":{"tenant_id":"acme","allowed_profiles":["default","remote"]}}'
# 或 OPENLCA_API_KEYS_FILE=config/api_keys.json
```
当身份验证开启时，发送 `Authorization: Bearer <key>`（流网关也接受
`?api_key=<key>`，适用于无法设置标头的客户端，如 ChatGPT）。
## 可用工具
所有 24 个工具都只在它们具有工作处理程序时才被宣传（在启动时和通过
回归测试强制执行），因此没有“死亡”工具。
### 连接和健康状况
| 工具 | 目的 | 输入 |
|------|---------|--------|
| `test_connection` | 测试 openLCA 连接 | 无 |
| `health_check` | 连接状态 + 实体计数 | count_entities? |
### 阶段 1：目标和范围定义
| 工具 | 目的 | 输入 |
|------|---------|--------|
| `search_flows` | 查找物料流动 | keywords, max_results?, flow_type? |
| `search_processes` | 查找工艺 | keywords, max_results? |
| `search_impact_methods` | 查找 LCIA 方法 (+ 类别) | keywords |
| `find_providers` | 查找流动的生产工艺 | flow_id 或 flow_name |
| `get_entity_by_name` | 精确名称查找 → 实体摘要 | model_type, name |
### 阶段 2：生命周期库存（LCI）— 写入
| 工具 | 目的 | 输入 |
|------|---------|--------|
| `create_product_flow` | 创建产品流动 | name, description? |
| `create_process` | 创建单元工艺 | name, description?, exchanges |
| `create_product_system` | 构建产品系统 | process_id 或 process_name |
> 当 `OPENLCA_READ_ONLY=true` 时，写入工具将返回 `WRITE_BLOCKED` 错误。
### 阶段 3：生命周期影响评估（LCIA）
| 工具 | 目的 | 输入 |
|------|---------|--------|
| `calculate_impacts` | 计算影响 → result_id + ResultSummary | system_id 或 system_name, method_id 或 method_keywords, amount? |
| `get_inventory_results` | 完整的 LCI（基本流动） | result_id, direction? |
| `get_total_requirements` | 缩放技术流动 | result_id |
| `get_normalized_impacts` | 归一化影响 | result_id |
| `get_weighted_impacts` | 加权影响 | result_id |
### 阶段 4：解释
| 工具 | 目的 | 输入 |
|------|---------|--------|
| `check_result_consistency` | 健全性检查结果（每类别贡献总和到总数） | result_id, rel_tol?, abs_tol? |
| `analyze_contributions` | 顶级工艺/流动贡献者 | result_id, impact_category_id, n?, contribution_type?, min_share? |
| `get_contribution_tree` | 上游热点树 | result_id, impact_category_id, max_depth?, min_share? |
| `get_sankey` | 桑基图数据 | result_id, impact_category_id, max_nodes?, min_share? |
| `compare_systems` | 在一个方法上比较两个系统 | system1_id, system2_id, method_id 或 method_keywords, amount? |
| `run_monte_carlo` | 不确定性分析（统计） | system_id, method_id 或 method_keywords, iterations? |
| `run_scenario_analysis` | 在值上变化参数 | system_id, method_id 或 method_keywords, parameter_name, values |
| `export_results` | 将影响/比较导出到 CSV/Excel | result_id 或 data, filepath, format?, kind? |
### 实用工具
| 工具 | 目的 | 输入 |
|------|---------|--------|
| `dispose_result` | 释放一个计算结果 | result_id |
| `dispose_all_results` | 释放所有跟踪的结果 | 无 |
### `result_id` 工作流
`calculate_impacts` 运行计算，存储实时结果，并返回一个 `result_id`
加上一个紧凑的 `ResultSummary` 和任何一致性警告。后续工具
（`analyze_contributions`, `get_contribution_tree`, `get_inventory_results`, `get_sankey`,
`get_normalized_impacts`, ...）接受该 `result_id`，以便结果被重用，而不是重新计算。
当完成时调用 `dispose_result`（所有结果在关闭时也会被释放）。
## 使用示例
### 示例 1：完整的 LCA 工作流（AI 代理视角）
AI 代理将按以下顺序调用工具：
```
1. test_connection()
→ 验证 openLCA 可访问
2. search_flows(keywords=["steel"])
→ 查找钢材流动 ID
3. find_providers(flow_id="...")
→ 查找钢材生产工艺 ID
4. create_product_flow(name="My Widget")
→ 创建产品流动，获取 ID
5. create_process(
name="Widget Production",
exchanges=[
{flow_id: "widget_id", amount: 1.0, is_input: false, is_quantitative_reference: true},
{flow_id: "steel_id", amount: 2.0, is_input: true, provider_id: "provider_id"}
]
)
→ 创建工艺，获取 ID
6. create_product_system(process_id="...")
→ 构建产品系统，获取 ID
7. search_impact_methods(keywords=["TRACI"])
→ 查找影响方法 ID
8. calculate_impacts(system_id="...", method_id="...")
→ 获取影响 + result_id
9. analyze_contributions(result_id="...", impact_category_id="...")
→ 确定热点
10. dispose_result(result_id="...")
→ 清理内存
```
### 示例 2：搜索和发现
```json
// 工具：search_flows
{
"keywords": ["polyethylene", "terephthalate"],
"max_results": 5,
"flow_type": "PRODUCT_FLOW"
}
// 响应：
{
"success": true,
"count": 2,
"flows": [
{
"id": "abc-123",
"name": "聚乙烯对苯二甲酸酯，颗粒，瓶级",
"category": "plastics"
}
]
}
```
### 示例 3：创建工艺
```json
// 工具：create_process
{
"name": "PET 瓶生产",
"description": "从颗粒生产 1 个 PET 瓶",
"exchanges": [
{
"flow_id": "bottle_flow_id",
"amount": 1.0,
"is_input": false,
"is_quantitative_reference": true
},
{
"flow_id": "pet_granulate_id",
"amount": 0.025,
"is_input": true,
"provider_id": "pet_production_id"
}
]
}
// 响应：
{
"success": true,
"process": {
"id": "process-123",
"name": "PET 瓶生产",
"description": "从颗粒生产 1 个 PET 瓶"
}
}
```
### 示例 4：计算影响
```json
// 工具：calculate_impacts
{
"system_id": "system-123",
"method_id": "traci-method-id",
"amount": 1.0
}
// 响应（缩略）：
{
"success": true,
"result_id": "res_a1b2c3d4e5f6",
"summary": {
"result_id": "res_a1b2c3d4e5f6",
"top_impacts": [
{"category": "全球变暖", "amount": 0.05, "unit": "kg CO2 eq"}
],
"next_actions": ["inspect_contribution_tree", "compare_scenario", "export_report"],
"warnings": []
},
"impacts": [
{"name": "全球变暖", "category": {"id": "gwp-id", "name": "全球变暖"},
"category_id": "gwp-id", "amount": 0.05, "unit": "kg CO2 eq"}
],
"warnings": [],
"context": { "result_id": "res_a1b2c3d4e5f6", "timestamp": "...", "package_version": "0.4.0" },
"message": "当完成时，使用此 result_id 调用 dispose_result。"
}
```
将 `impacts[].category_id` 馈入 `analyze_contributions` / `get_contribution_tree` /
`get_sankey` 作为 `impact_category_id`。
## n8n 集成
### 设置步骤
1. **在 n8n 中安装 MCP 集成**
- 在您的 n8n 实例中安装 MCP 节点
- 配置 MCP 连接
2. **配置 MCP 服务器连接**（可流式 HTTP）
```json
{
"name": "OpenLCA LCA 服务器",
"serverType": "http",
"url": "http://localhost:8000/mcp"
}
```
> 服务器在 `/mcp` 处使用 **可流式 HTTP**。在流网关后面
> （`docker-compose.gateway.yml`）公共端点是 `https://<host>/mcp`。
3. **创建 n8n 工作流**
参见 [examples/n8n-workflows/](examples/n8n-workflows/) 了解完整的工作流模板。
### 4 阶段 LCA 工作流模板
您的 n8n 工作流应具有 4 个代理阶段：
**阶段 1 代理：目标和范围**
- 使用：`test_connection`, `search_flows`, `search_processes`, `search_impact_methods`
- 输出：所需材料、工艺和影响方法的列表
**阶段 2 代理：生命周期库存**
- 使用：`create_product_flow`, `create_process`, `create_product_system`
- 输出：产品系统 ID 准备计算
**阶段 3 代理：影响评估**
- 使用：`calculate_impacts`, `get_inventory_results`
- 输出：具有 result_id 的影响结果
**阶段 4 代理：解释**
- 使用：`analyze_contributions`, `export_results`, `dispose_result`
- 输出：分析报告和导出文件
参见 [docs/n8n-integration.md](docs/n8n-integration.md) 了解详细设置。
## 架构
```
┌─────────────────────────────────────────┐
│ n8n 工作流代理 │
│ ┌────────┬────────┬────────┬────────┐ │
│ │阶段 1 │阶段 2 │阶段 3 │阶段 4 │ │
│ │ 目标和范围 │ 生命周期库存 │ 生命周期影响评估 │ 解释 │ │
│ │ 范围 │ │ │ │ │
│ └───┬────┴───┬────┴───┬────┴───┬────┘ │
│ │ │ │ │ │
│ └────────┴────────┴────────┘ │
│ MCP 协议 │
└──────────────────┬──────────────────────┘
│
┌────────▼────────┐
│ MCP 服务器 │
│ (此服务器) │
└────────┬────────┘
│
┌────────▼────────┐
│ openlca-ipc 库 │
└────────┬────────┘
│
┌────────▼────────┐
│ openLCA IPC │
│ 服务器 │
└────────┬────────┘
│
┌────────▼────────┐
│openLCA 桌面版 │
│ 应用程序 │
└─────────────────┘
```
## 错误处理
每个响应都带有 `success` 布尔值。失败还附带结构化
代理字段，以便 AI 代理可以编程分支：
```json
// 成功响应
{
"success": true,
"...": "工具特定有效载荷"
}
// 错误响应
{
"success": false,
"is_error": true,
"error_code": "IMPACT_METHOD_NOT_FOUND",
"message": "没有匹配的冲击方法 ['nope']。",
"recoverable": true,
"suggested_next_actions": ["search_impact_methods"]
}
```
错误代码包括 `CONNECTION_FAILED`, `IMPACT_METHOD_NOT_FOUND`, `SYSTEM_NOT_FOUND`,
`ENTITY_NOT_FOUND`, `CALCULATION_FAILED`, `WRITE_BLOCKED`, `UNKNOWN_TOOL`, 和
`INTERNAL_ERROR`。
| 错误代码 | 原因 | 建议恢复 |
|-----------|-------|--------------------|
| `CONNECTION_FAILED` | IPC 服务器未运行 | 在 openLCA 中启动 IPC 服务器 |
| `ENTITY_NOT_FOUND` | 未找到流程/结果 | 尝试不同的关键词；验证 ID |
| `SYSTEM_NOT_FOUND` | 没有匹配的产品系统 | 搜索或创建产品系统 |
| `IMPACT_METHOD_NOT_FOUND` | 没有匹配的 LCIA 方法 | 使用 `search_impact_methods` |
| `WRITE_BLOCKED` | 在只读模式下尝试写入 | 设置 `OPENLCA_READ_ONLY=false` |
## AI 代理最佳实践
### 1. 始终先测试连接
```
步骤 1：调用 test_connection()
步骤 2：如果连接，继续工作流
```
### 2. 检查搜索结果
```
步骤 1：调用 search_flows(keywords=[...])
步骤 2：如果计数 > 0，使用第一个结果
步骤 3：如果计数 == 0，尝试替代关键词
```
### 3. 始终释放结果
```
步骤 1：调用 calculate_impacts() → 获取 result_id
步骤 2：使用 result_id 进行分析
步骤 3：完成时调用 dispose_result(result_id)
```
### 4. 宽容处理缺失数据
```
如果搜索返回空：
- 尝试更广泛的关键词
- 尝试替代拼写
- 创建所需数据
- 报告给用户
```
## 开发
### 开发模式运行
```bash
# 启用调试日志
export LOG_LEVEL=DEBUG
# 运行服务器
python -m src
```
### 测试工具
使用 MCP 检查器或直接工具调用：
```bash
# 安装 MCP 检查器
npm install -g @modelcontextprotocol/inspector
# 检查服务器
mcp-inspector python -m src
```
服务器分为专注的模块；`src/server.py` 仍然是入口点：
| 模块 | 责任 |
|--------|----------------|
| `src/server.py` | 服务器设置、调度、stdio + 可流式 HTTP + 旧版 SSE 传输、 `main` / `run` |
| `src/tool_defs.py` | `TOOLS` — 所有工具模式定义 |
| `src/handlers.py` | 异步处理程序 + `TOOL_HANDLERS` 映射 |
| `src/lca_client.py` | `get_client()` 单例、环境配置、只读模式 |
| `src/result_store.py` | `result_id` 注册表，用于实时结果 |
| `src/responses.py` | 成功/错误信封 + 序列化器 |
要添加工具：
1. 在 `src/tool_defs.py` 中注册其模式（使用 `_register(_tool(...))` 帮助器）。
2. 在 `src/handlers.py` 中实现 `async def handle_<name>(arguments)` 并将其添加到
`TOOL_HANDLERS`。
`TOOLS` 和 `TOOL_HANDLERS` 必须保持同步 — 服务器在启动时引发错误，测试
失败，否则会报错，因此已宣传的工具永远不会缺少处理程序。
## 故障排除
### 服务器无法启动
**问题：** 服务器无法启动
**检查：**
1. 是否安装了 Python 3.11+？ `python --version`
2. 是否已安装依赖项？ `pip list | grep -E "mcp|openlca-ipc"`
3. 是否配置了 `.env` 文件？ `cat .env`
### 无法连接到 openLCA
**问题：** `test_connection` 返回 false
**检查：**
1. openLCA 是否正在运行？
2. 数据库是否已加载？
3. IPC 服务器是否已启动？
4. `.env` 文件中的端口是否与 IPC 服务器匹配？
### 工具调用失败
**问题：** 工具返回错误
**检查：**
1. 查看响应中的错误消息
2. 检查服务器日志（LOG_LEVEL=DEBUG）
3. 验证输入是否与模式匹配
4. 使用 MCP 检查器进行测试
## 日志
查看日志来调试问题：
```bash
# 设置日志级别
export LOG_LEVEL=DEBUG
```
# 运行服务器和查看日志
python -m src 2>&1 | tee server.log
```
日志级别：
- **调试**: 所有操作和数据
- **信息**: 主要操作（默认）
- **警告**: 不影响运行的问题
- **错误**: 失败和异常
## 性能
### 资源使用
- **内存**: 基线约50-100 MB
- **CPU**: 最小（大部分工作在openLCA中）
- **网络**: 仅限本地主机（进程间通信）
### 优化建议
1. **及时释放结果** - 防止内存泄漏
2. **限制搜索结果** - 使用max_results参数
3. **缓存影响方法** - 重用方法ID
4. **批量操作** - 在计算前创建多个进程
## 安全
### 仅限本地使用
该服务器设计为**仅限本地使用**：
- 连接到本地openLCA
- 无网络暴露
- 无身份验证要求
### 生产部署
用于生产环境时：
- 添加身份验证
- 限制工具访问
- 审计工具调用
- 限制并发连接
### 远程访问和可选身份验证令牌
当您通过流式网关（`docker-compose.gateway.yml`）将服务器暴露给远程客户端（ChatGPT，Claude）时，身份验证是**可选的**：
- **无令牌（最简单）**：在`.env`中留空`MCP_AUTH_TOKEN`。网关运行开放 - 任何可以访问您的隧道/VPN的人都可以调用它。当隧道本身已经限制访问时，使用裸URL：`https://<your-host>/mcp`
- **使用令牌**：将`MCP_AUTH_TOKEN`设置为一个长随机密钥。网关随后在每个请求中都需要它。由于ChatGPT的连接器UI无法发送自定义标头，请作为查询参数传递：`https://<your-host>/mcp?api_key=<MCP_AUTH_TOKEN>`
**生成令牌**（任意一个），然后粘贴到`.env`中的`MCP_AUTH_TOKEN`：
```bash
# macOS / Linux（或Windows上的Git Bash）
openssl rand -hex 32
# Windows PowerShell
python -c "import secrets; print(secrets.token_hex(32))"
```
```env
# .env — 留空以运行开放网关，或在此处粘贴生成的密钥
MCP_AUTH_TOKEN=
```
> 令牌**仅**由网关（`gateway/Caddyfile`）强制执行。在stdio模式下或在没有网关的情况下运行裸服务器时，它不起作用。有关完整的远程设置，请参见[docs/online-hosting.md](docs/online-hosting.md)。
## 贡献
欢迎贡献！请参阅主项目[CONTRIBUTING.md](../CONTRIBUTING.md)。
## 许可证
MIT许可证 - 有关详细信息，请参见[LICENSE](../LICENSE)。
## 支持
- **文档**：[完整文档](docs/)
- **问题**：[GitHub 问题](https://github.com/SDAI-institute/openlca-mcp/issues)
- **电子邮件**：dernestbanksch@gmail.com
## 致谢
基于：
- [MCP协议](https://modelcontextprotocol.io/)，由Anthropic提供
- [openlca-ipc](https://github.com/SDAI-institute/openlca-ipc)
- [openLCA](https://www.openlca.org/)，由GreenDelta提供
---
**准备好使用AI代理自动执行LCA工作流程！**
MCP 配置
以下是此 MCP Server 的配置信息，可直接复制到 Cursor 或其他 MCP 客户端中使用。
mcp.json
连接信息
猜你喜欢
暂无推荐服务器