工作流调试服务器

Temporal MCP Server是一个用于调试Temporal工作流的服务器,通过提供工作流检查和历史事件查询功能,支持LLM辅助调试。

  • by kk-kd
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 kk-kd/temporal-mcp

使用说明

临时MCP服务器

用于调试时态工作流的MCP(模型上下文协议)服务器。此服务器通过公开Temporal工作流检查工具启用LLM辅助调试。

特性

  • list_workflows -使用可选查询筛选器列出工作流执行情况
  • 描述工作流 -获取特定工作流的详细信息
  • get_workflow_history -获取事件历史记录以进行调试

需求

  • Python 3.11+
  • 紫外线 用于依赖关系管理
  • 正在运行的临时服务器(本地或云)

安装

# Clone the repository
git clone <repo-url>
cd temporal-mcp

# Create virtual environment and install
uv venv
source .venv/bin/activate
uv pip install -e ".[dev]"

配置

设置环境变量以配置时态连接:

变量 描述 默认值
TEMPORAL_ADDRESS 临时服务器地址 localhost:7233
TEMPORAL_NAMESPACE 时态命名空间 default
TEMPORAL_TLS_CERT TLS证书的路径(适用于云) -
TEMPORAL_TLS_KEY TLS密钥路径(适用于云) -
TEMPORAL_API_KEY API密钥(用于云) -

码头工人

连接到临时服务器

如果您正在运行Temporal(本地或远程):

# Connect to Temporal on host machine
docker compose up

# Or specify a custom address
TEMPORAL_ADDRESS=my-temporal:7233 docker compose up

连接到Temporal Cloud

对于具有mTLS证书的Temporal Cloud:

# Place your certificates in a certs/ directory, then:
docker compose up
# After uncommenting the TLS environment variables in docker-compose.yml

或者使用API密钥:

TEMPORAL_ADDRESS=your-ns.tmprl.cloud:7233 \
TEMPORAL_NAMESPACE=your-ns \
TEMPORAL_API_KEY=your-key \
docker compose up

本地开发(无Docker)

直接用Python安装并运行:

对于Temporal Cloud,设置适当的环境变量:

export TEMPORAL_ADDRESS="your-namespace.tmprl.cloud:7233"
export TEMPORAL_NAMESPACE="your-namespace"
export TEMPORAL_API_KEY="your-api-key"

或者使用TLS证书:

export TEMPORAL_ADDRESS="your-namespace.tmprl.cloud:7233"
export TEMPORAL_NAMESPACE="your-namespace"
export TEMPORAL_TLS_CERT="/path/to/cert.pem"
export TEMPORAL_TLS_KEY="/path/to/key.pem"

用法

带光标

添加到光标MCP设置(~/.cursor/mcp.json):

{
  "mcpServers": {
    "temporal": {
      "command": "uv",
      "args": ["run", "temporal-mcp"],
      "cwd": "/path/to/temporal-mcp",
      "env": {
        "TEMPORAL_ADDRESS": "localhost:7233",
        "TEMPORAL_NAMESPACE": "default"
      }
    }
  }
}

独立

temporal-mcp

可用工具

list_workflows

列出带有可选筛选的工作流执行情况。

list_workflows(query="", limit=10)
  • query:可选的时态列表过滤器语法(例如。, WorkflowType="MyWorkflow" AND ExecutionStatus="Running")
  • limit:要返回的最大工作流数(默认值10,最大值100)

描述工作流

获取特定工作流的详细信息。

describe_workflow(workflow_id, run_id="")
  • workflow_id:要描述的工作流ID
  • run_id:可选运行ID(如果未指定,则使用最新的)

get_workflow_history

获取工作流执行的事件历史记录。

get_workflow_history(workflow_id, run_id="", max_events=100)
  • workflow_id:工作流ID
  • run_id:可选运行ID(如果未指定,则使用最新的)
  • max_events:要返回的最大事件数(默认值100,最大值1000)

发展

设置

uv venv
source .venv/bin/activate
uv pip install -e ".[dev]"

命令

# Format code
black src tests

# Lint
ruff check src tests

# Type check
mypy src

# Run tests
pytest

# Run all checks
black src tests && ruff check src tests && mypy src && pytest

项目结构

temporal-mcp/
├── pyproject.toml          # Project config and dependencies
├── Dockerfile
├── docker-compose.yml       # Docker setup for MCP server
├── README.md
├── src/temporal_mcp/
│   ├── __init__.py
│   ├── server.py           # MCP server and tool definitions
│   ├── client.py           # Temporal client wrapper
│   ├── config.py           # Environment-based configuration
│   └── models.py           # Pydantic models
├── tests/
│   ├── conftest.py         # Test fixtures
│   ├── test_config.py
│   ├── test_client.py
│   └── test_server.py
└── docs/
    └── PLAN.txt

许可证

麻省理工学院