MCP Project Workflows

MCP工作流生成器系统-完整指南

什么是MCP工作流生成器？

MCP（模型上下文协议）工作流生成器是一个 生产就绪的Python框架 用于创建、管理和执行结合了机器学习、自动化和人在环过程的复杂工作流。它专为您需要的场景而设计 确定性、可审计的工作流程 它可以集成AI模型（LLM）、自动化脚本和手动步骤。

核心目的

• 记笔记和记忆：持久内存文件跟踪工作流执行的进度
• 结构化开发：对复杂任务（代码开发、文档编制、研究）实施系统方法
• AI+自动化：将LLM功能与shell脚本、Python自动化和手动监督相结合
• 可重用性：为常见工作流创建模板（设计→ 实施→ 测试模式）

架构概述

关键组件

1. 工作流规范 -不可变的心

定义完整工作流的冻结数据类：

@dataclass(frozen=True)
class WorkflowSpec:
    goal: str                    # "Write production-ready code for task X"
    memory_file: str            # "memory.md" - persistent context
    tasks: tuple[TaskSpec, ...] # Reusable documents/instructions
    steps: tuple[StepSpec, ...] # Executable workflow units

2. 工作流生成器 -Fluent建筑API

使用可链式方法逐步构建工作流：

WorkflowBuilder.start() \
    .with_goal("Build a web app") \
    .memory("memory.md") \
    .register_task("requirements", file="tasks/reqs.md") \
    .add_step("Design", kind="llm", uses=["requirements"]) \
    .compile()

graph LR
    A["WorkflowBuilder.start()"] --> B["with_goal('...')"]
    B --> C["memory('...')"]
    C --> D["register_task('...', file='...')"]
    D --> E["add_step('Design', kind='llm', uses=[...])"]
    E --> F["register_task('...', file='...')"]
    F --> G["add_step('Implement', ...)"]
    G --> H["compile()"]
    H --> I["WorkflowSpec"]

3. 工作流编排器 -执行引擎

逐步执行已编译的工作流：

orchestrator = WorkflowOrchestrator(spec)
responses = orchestrator.run()  # Returns StepResponse list

graph TD
    A["WorkflowOrchestrator"] --> B["Load Memory File"]
    B --> C["For Each Step in WorkflowSpec"]
    C --> D["Create StepRequest"]
    D --> E["Resolve Executor by StepKind"]
    E --> F["Executor.execute(request)"]
    F --> G{"Response Status?"}
    G -->|OK| H["Notify Observer.on_finish"]
    G -->|FAIL| I["Notify Observer.on_error"]
    G -->|RETRY| J["Handle Retry Logic"]
    H --> K["Append Result to Memory"]
    I --> K
    K --> L{"Has Next Step?"}
    J --> C
    L -->|Yes| C
    L -->|No| M["Return All StepResponses"]

4. 执行器 -可插拔的执行策略

执行步骤的不同方法：

• LLM：使用语言模型（目前返回预设响应）
• 外壳：执行shell命令
• python：运行Python脚本

真实世界使用示例

基于你的 my_test_workflow/workflow.yaml:

goal: Write production-ready code for the specified task
memory_file: memory.md
tasks:
- id: requirements
  text: "# Detailed requirements gathering instructions..."
steps:
- id: 1
  name: Gather Requirements
  kind: llm
  uses: [requirements]  # References the requirements task

这创建了一个 基于指导的工作流 其中每个步骤都将相关文档作为上下文注入。

任务和步骤如何协同工作

graph TD
    A[Task: requirements] --> B[Markdown content with requirements guidelines]
    C[Task: design] --> D[Design principles and approach]
    E[Task: implement] --> F[Coding standards and best practices]

    B --> G[Step 1: Gather Requirements]
    G --> H[uses requirement task content as context]

    D --> I[Step 2: Design Solution]
    I --> J[uses design + requirements context]

    F --> K[Step 3: Implement Code]
    K --> L[uses implement + design + requirements]

如何使用MCP工作流生成器

快速入门-使用模板

from mcp_workflows.templates import get_template, create_workflow_from_template

# Use a predefined template (code_workflow includes requirements/design/implement/test)
template = get_template("code_workflow")
workflow_path = create_workflow_from_template(template, Path("workflows/my_app"))

手动生成器模式

from mcp_workflows.builder import WorkflowBuilder
from mcp_workflows.spec import StepKind
from pathlib import Path

builder = WorkflowBuilder.start()

# Basic configuration
builder.with_goal("Create a React component library")
builder.memory("workflows/component_lib/memory.md")

# Register reusable documentation
builder.register_task(
    "design_principles",
    text="# Design Principles\n- Use TypeScript\n- Follow atomic component patterns..."
)

builder.register_task(
    "component_specs",
    file="workflows/component_lib/specs/component_specs.md"
)

# Define steps
builder.add_step(
    name="Design Architecture",
    kind=StepKind.LLM,
    doc="Design the component library structure",
    uses=["design_principles"],
    input_template="Design components for: {feature_name}",
    config={"temperature": 0.3}
)

builder.add_step(
    name="Create Components",
    kind=StepKind.LLM,
    doc="Implement the designed components",
    uses=["component_specs", "design_principles"],
    config={"model": "gpt-4"}
)

# Compile and save
spec = builder.compile()
builder.emit_yaml("workflows/component_lib/workflow.yaml")

执行工作流

from mcp_workflows.orchestrator import WorkflowOrchestrator

# Load compiled spec
orchestrator = WorkflowOrchestrator(spec)

# Optional: Add observers for monitoring
class WorkflowMonitor:
    def on_step_start(self, request):
        print(f"Starting: {request.name}")

    def on_step_finish(self, request, response):
        print(f"Completed: {request.name} -> {response.status}")

orchestrator_with_monitoring = WorkflowOrchestrator(
    spec,
    observer=WorkflowMonitor()
)

# Execute
responses = orchestrator.run()
print(f"Workflow completed with {len(responses)} steps")

深潜：了解每个组件

任务系统：可重用性和上下文

任务与步骤：

• 任务：可重用的文档/定义（如可以多次调用的函数）
• 步骤：引用任务的单个执行实例

任务类型：

1. 基于文件: register_task("api_docs", file="docs/api.md")
2. 基于文本: register_task("instructions", text="Step-by-step guide...")

高级任务使用情况：

# Multi-context steps
builder.add_step(
    name="Code Review",
    kind="llm",
    uses=["requirements", "design", "implementation"],  # 3 different contexts
    input_template="Review this code against requirements and design"
)

步骤配置深度学习

步骤规格字段：

• id：顺序整数标识符
• name：描述性名称
• kind：执行策略（llm/shell/python）
• doc：文档（为什么存在此步骤）
• uses：任务ID列表（上下文文档）
• input_template：动态输入格式
• config：执行器特定参数
• branches：有条件跳跃
• next_step：自定义订单覆盖

输入模板：

# Templates with variable substitution
input_template="Analyze {feature_name} for {target_platform}"

# Under the hood, gets formatted with step config
formatted = "Analyze authentication for mobile platform"

分支逻辑：

builder.add_step(
    name="Code Review",
    # ... other fields
    branches=[
        Branch(when="failed linting", goto=5),  # Jump to fix step
        Branch(when="tests failed", goto=7)     # Jump to test fixing
    ]
)

存储系统：持久上下文

存储的内容：

• 步骤执行摘要
• 错误消息（如有）
• 关键工件/输出摘录

内存格式：

- Gather Requirements: Complete requirements gathered for user auth
- Design Solution: Designed JWT-based authentication with refresh tokens
- Implement Code: Created Login component, UserContext, ProtectedRoute
- Test and Review: Tests passing, code quality good

记忆持久性的好处：

• 恢复工作流：中断后重新启动
• 语境意识：未来的步骤看看做了什么
• 可审计历史：完整的执行轨迹
• 调试：易于发现故障发生的位置

定制与扩展

创建自定义模板

from mcp_workflows.templates import WorkflowTemplate
from mcp_workflows.spec import StepKind

class DataScienceTemplate(WorkflowTemplate):
    name = "data_science"

    def __init__(self):
        super().__init__()
        self.goal = "Build and deploy ML model for prediction task"
        self.steps = [
            StepSpec(
                id=1, name="Data Exploration",
                kind=StepKind.PYTHON,
                doc="Analyze dataset characteristics"
            ),
            StepSpec(
                id=2, name="Feature Engineering",
                kind=StepKind.LLM,
                doc="Design features for model input"
            ),
            StepSpec(
                id=3, name="Model Building",
                kind=StepKind.PYTHON,
                doc="Train and evaluate models"
            )
        ]

模板创建流程

graph TD
    A[Extend WorkflowTemplate] --> B[Define name]
    B --> C[Set goal]
    C --> D[Configure memory_file]
    D --> E[Define base_tasks]
    E --> F[Implement __post_init__]

    F --> G[Create StepSpec list]
    G --> H[User calls create_workflow_from_template]
    H --> I[Template converts BaseTasks to TaskSpecs]
    I --> J[template.tasks property]
    J --> K[Builder creates WorkflowSpec]

自定义执行器

from mcp_workflows.executors import Executor
from mcp_workflows.spec import StepRequest, StepResponse

class RESTAPIExecutor(Executor):
    def execute(self, request: StepRequest) -> StepResponse:
        # Execute HTTP API calls
        response = requests.post(f"http://api.example.com/{request.name}")
        return StepResponse(
            status="ok" if response.ok else "fail",
            result=response.json(),
            error=response.text if not response.ok else None
        )

# Register custom executor
factory = ExecutorFactory.default()
factory.register_instance(StepKind.LLM, CustomLLMExecutor())

高级依赖注入

from mcp_workflows.factories import ServiceContainer, ExecutorFactory

# Custom container with external services
container = ServiceContainer()

# Register singletons
container.register_singleton("api_client", lambda _: requests.Session())
container.register_singleton(
    "llm_service",
    lambda c: OpenAIClient(api_key=os.env["OPENAI_API_KEY"])
)

# Create factory and register custom executors
factory = ExecutorFactory(container)
factory.register_factory(
    "ml_prediction",
    lambda c: MLExecutor(c.resolve("api_client"))
)

依赖注入流程

graph TD
    A[ServiceContainer] --> B[register_singleton]
    B --> C[Store factory function]
    D[ExecutorFactory] --> E[create]
    E --> F[Container.resolve]
    F --> G{Cached singleton?}
    G -->|Yes| H[Return cached instance]
    G -->|No| I[Call factory function]
    I --> J[Cache singleton]
    J --> H

提示、技巧和最佳实践

1.内存管理

# Custom memory formatter for better context
def summarize_step(name: str, response: StepResponse) -> str:
    if response.status == "ok":
        # Extract key insights
        return f"- {name}: ✓ {response.result.get('key_outcome', 'completed')}"
    else:
        return f"- {name}: ✗ Failed - {response.error}"

2.模板继承

class SpecializedCodeTemplate(CodeWorkflowTemplate):
    """Extend base template with specific steps"""

    def __init__(self, tech_stack: str):
        super().__init__()
        self.steps.append(
            StepSpec(
                id=5,
                name=f"Tech Stack Setup ({tech_stack})",
                kind=StepKind.SHELL,
                doc=f"Initialize {tech_stack} project structure"
            )
        )

3.有条件的工作流

def build_deployment_workflow(env: str):
    builder = WorkflowBuilder.start().with_goal("Deploy application")

    steps = [
        ("build", StepKind.SHELL, "Build application"),
        ("test", StepKind.PYTHON, "Run test suite"),
    ]

    if env == "production":
        steps.extend([
            ("staging_deploy", StepKind.SHELL, "Deploy to staging"),
            ("integration_test", StepKind.PYTHON, "Verify staging"),
        ])

    steps.append(("production_deploy", StepKind.SHELL, "Deploy to production"))

    for name, kind, doc in steps:
        builder.add_step(name=name, kind=kind, doc=doc)

    return builder.compile()

4.用于监控的观察者模式

class ComprehensiveMonitor:
    def on_step_start(self, request: StepRequest):
        logger.info(f"[{request.correlation_id}] Starting {request.name}")

    def on_step_finish(self, request: StepRequest, response: StepResponse):
        if response.result and 'artifacts' in response.result:
            # Save artifacts to persistent storage
            save_artifacts(request.name, response.result['artifacts'])

    def on_step_error(self, request: StepRequest, response: StepResponse):
        # Send notifications
        send_notification(f"Step failed: {request.name} - {response.error}")
        # Auto-retry logic
        if should_retry(response.error):
            return RetryDecision(retries_left=3)

观察者生命周期图

sequenceDiagram
    participant W as WorkflowOrchestrator
    participant O as Observer
    participant E as Executor
    participant M as Memory

    W->>O: on_step_start(request)
    W->>E: execute(request)
    E-->>W: StepResponse
    alt Status OK
        W->>O: on_step_finish(request, response)
        W->>M: append_result()
    else Status FAIL
        W->>O: on_step_error(request, response)
        W->>M: append_error()
    else Status RETRY
        W->>O: on_step_error(request, response)
        W->>W: retry_logic()
    end

5.并行子工作流程

def create_parallel_validation_workflow():
    """Run validation steps in parallel (simulate)"""

    parallel_steps = [
        ("security_audit", StepKind.PYTHON),
        ("performance_test", StepKind.SHELL),
        ("accessibility_check", StepKind.LLM),
    ]

    # Execute each parallel step
    # Note: Current system is sequential, but this pattern shows extensibility
    for step_name, step_kind in parallel_steps:
        builder.add_step(name=f"Parallel {step_name}", kind=step_kind)

6.配置驱动的工作流

@dataclass
class WorkflowConfig:
    name: str
    steps: list[dict]
    memory_strategy: str = "accumulate"

def build_from_config(config: WorkflowConfig):
    builder = WorkflowBuilder.start().with_goal(f"Execute {config.name}")

    for step_config in config.steps:
        builder.add_step(**step_config)

    return builder.compile()

常见模式和用例

1.研究助理工作流程

1. Literature Review (LLM + Web)
2. Hypothesis Formation (LLM)
3. Experiment Planning (LLM)
4. Data Collection (Python)
5. Analysis (Python)
6. Results Interpretation (LLM)

2.代码开发管道

1. Requirements Analysis (LLM)
2. Architecture Design (LLM)
3. Implementation (LLM/Programming)
4. Unit Testing (Python)
5. Integration Testing (Shell)
6. Documentation (LLM)

3.内容创建管道

1. Topic Research (LLM)
2. Content Planning (LLM)
3. Draft Creation (LLM)
4. Fact Checking (LLM)
5. Editing & Polish (LLM)
6. SEO Optimization (LLM)

4.业务流程自动化

1. Data Import (Shell)
2. Data Validation (Python)
3. Report Generation (LLM + Data)
4. Approval Routing (Custom Logic)
5. Distribution (Shell/Email)

故障排除指南

常见问题

1.未找到执行人

# Problem: StepKind.LLM executor not registered
# Solution: Use ExecutorFactory.default() or register explicitly

2.内存文件竞争条件

# Problem: Concurrent workflow access
# Solution: Use file locking or queue-based execution
import fcntl
with open(memory_file, 'a') as f:
    fcntl.flock(f.fileno(), fcntl.LOCK_EX)

3.长时间运行步骤超时

# Problem: Steps taking too long
# Solution: Add timeout configuration
builder.add_step(
    name="Long Process",
    kind="shell",
    config={"timeout": 300}  # 5 minutes
)

4.内存文件大小爆炸

# Problem: Memory file growing too large
# Solution: Implement memory rotation or summarization

调试模式

class DebugMonitor:
    def on_step_start(self, request):
        print(f"DEBUG: Request: {request}")
        print(f"DEBUG: Input: {request.input}")
        print(f"DEBUG: Config: {request.config}")

    def on_step_finish(self, request, response):
        print(f"DEBUG: Response: {response}")
        print(f"DEBUG: Memory updated: {response.status}")

故障排除决策流程

flowchart TD
    A[Workflow Step Fails] --> B{Error Type?}
    B -->|Execution Timeout| C[Configure step timeouts]
    B -->|Command Injection| D[Use args list instead of string]
    B -->|Environment Missing| E[Check ENV variables set]
    B -->|Network Failure| F[Add retry logic]
    B -->|File Permission| G[Verify path permissions]
    B -->|Memory Corruption| H[Check concurrent access]

    C --> I[Log detailed error info]
    D --> I
    E --> I
    F --> I
    G --> I
    H --> I

    I --> J[Enable Debug Monitor]
    J --> K[Check request/response data]
    K --> L[Validate executor configuration]
    L --> M[Verify task/step relationships]
    M --> N[Test with minimal example]

性能优化

1.执行器池

# Reuse expensive resources
factory.register_singleton(
    StepKind.LLM,
    lambda: CachedLLMExecutor(model_cache_size=10)
)

2.分步批处理

# Group similar steps
# Implementation needed: BatchExecutor that processes multiple steps at once

3.内存优化

# Clear old memory for long workflows
def compress_memory(memory_text: str, max_lines: int = 1000) -> str:
    lines = memory_text.split('\n')
    if len(lines) > max_lines:
        # Keep recent lines + summary of older ones
        return summarize_old_lines(lines[:-max_lines]) + '\n'.join(lines[-max_lines:])
    return memory_text

安全注意事项

1.文件系统访问

# Sandbox file operations
import os
import pathlib

def safe_file_path(user_path: str, allowed_dir: str) -> pathlib.Path:
    resolved = pathlib.Path(allowed_dir) / user_path
    resolved = resolved.resolve()
    if not str(resolved).startswith(allowed_dir):
        raise ValueError("Path traversal attempt")
    return resolved

2.命令注入预防

# Use argument lists, not string concatenation
subprocess.run(["bash", "-c", "safe command"], shell=False)

3.API密钥管理

# Environment-based secrets
import os
api_key = os.environ.get("SECURE_API_KEY")
if not api_key:
    raise ValueError("API key required")

高级功能和扩展

1.自定义步骤类型

from mcp_workflows.spec import StepKind

try:
    # Register new kind
    StepKind.DATABASE = "database"
    StepKind.NOTIFICATION = "notification"
except:
    # If enum modification fails, create new types
    pass

2.工作流组成

class WorkflowComposer:
    def __init__(self):
        self.workflows = {}

    def register(self, name: str, spec: WorkflowSpec):
        self.workflows[name] = spec

    def compose(self, names: list[str]) -> WorkflowSpec:
        """Merge multiple workflows into one"""
        # Implementation for workflow composition

3.动态步长生成

def generate_steps_from_config(config_data: dict) -> list[StepSpec]:
    """Generate steps from configuration data"""
    steps = []
    for i, step_config in enumerate(config_data['steps'], 1):
        steps.append(StepSpec(
            id=i,
            name=step_config['name'],
            kind=StepKind(step_config['type']),
            config=step_config.get('config', {})
        ))
    return steps

4.实时监控GUI

# Integration with web frameworks for dashboards
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/workflow/status')
def get_status():
    return jsonify(active_workflows)

真实世界的例子

示例1：自动代码审查

builder = WorkflowBuilder.start() \
    .with_goal("Comprehensive code review") \
    .register_task("code_quality", file="standards.md") \
    .register_task("security_guidelines", file="security.md")

builder.add_step("Style Check", StepKind.SHELL, uses=["code_quality"])
builder.add_step("Security Scan", StepKind.PYTHON, uses=["security_guidelines"])
builder.add_step("Logic Review", StepKind.LLM, uses=["code_quality", "security_guidelines"])

spec = builder.compile()

示例2：数据管道

def create_etl_workflow(source: str, target: str):
    return WorkflowBuilder.start() \
        .with_goal(f"ETL from {source} to {target}") \
        .add_step("Extract", StepKind.PYTHON, config={"source": source}) \
        .add_step("Transform", StepKind.PYTHON, config={"transform_rules": "..."}) \
        .add_step("Load", StepKind.PYTHON, config={"target": target}) \
        .add_step("Validate", StepKind.PYTHON) \
        .compile()

示例3：多模式AI工作流

builder.add_step(
    "Image Analysis",
    kind="llm",
    config={"model": "gpt-4-vision"},
    uses=["analysis_guidelines"]
)

builder.add_step(
    "Text Summarization",
    kind="llm",
    config={"model": "claude-2"},
    uses=["summarization_template"]
)

未来扩展和路线图

计划的功能

• 并行执行：当依赖关系允许时，同时运行步骤
• 条件步骤评估：根据以前的结果跳过步骤
• 子工作流调用：将其他工作流作为步骤调用
• 事件流：实时进度通知
• 配置热重新加载：修改工作流而不重新启动
• 版本控制集成：跟踪工作流程随时间的变化

插件架构

# Planned plugin system
from mcp_workflows import plugin

@plugin.register_executor("slack")
class SlackExecutor(Executor):
    def execute(self, request):
        # Send Slack notifications
        pass

MCP工作流生成器提供 结构化、可扩展的框架 用于创建结合人工智能、自动化和人工监督的复杂工作流程。主要优势：

• 不可变规格：线程安全、可测试的工作流定义
• 持久存储器：在执行过程中保持上下文
• 可插拔执行器：易于添加新的执行类型
• 模板系统：可重用的工作流模式
• 观察者模式：丰富的监测和干预点

无论您是在构建人工智能辅助开发工具、自动化研究管道还是业务流程自动化，MCP Workflow Builder都为 可靠、可审计和可扩展的工作流系统.