MCP 与工作流引擎（如 Airflow）

# 示例：MCP Airflow Operator 实现代码
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from mcp.v2.client import MCPHTTPClient
from mcp.v2.exceptions import MCPException

class MCPOperator(BaseOperator):
    """
    MCP Airflow Operator，用于在 Airflow 工作流中调用 MCP 工具
    
    :param mcp_conn_id: MCP 连接 ID
    :param tool_name: 要调用的 MCP 工具名称
    :param tool_params: 工具调用参数
    :param mcp_timeout: MCP 调用超时时间（秒）
    :param mcp_retry_count: MCP 调用重试次数
    :param mcp_retry_delay: MCP 调用重试延迟（秒）
    :param xcom_push: 是否将 MCP 执行结果推送到 XCom
    :param xcom_key: XCom 键名，默认为 task_id
    """
    
    template_fields = ('tool_params',)
    
    @apply_defaults
    def __init__(
        self,
        mcp_conn_id: str = 'mcp_default',
        tool_name: str = None,
        tool_params: dict = None,
        mcp_timeout: int = 300,
        mcp_retry_count: int = 3,
        mcp_retry_delay: int = 10,
        xcom_push: bool = True,
        xcom_key: str = None,
        *args, **kwargs
    ):
        super().__init__(*args, **kwargs)
        self.mcp_conn_id = mcp_conn_id
        self.tool_name = tool_name
        self.tool_params = tool_params or {}
        self.mcp_timeout = mcp_timeout
        self.mcp_retry_count = mcp_retry_count
        self.mcp_retry_delay = mcp_retry_delay
        self.xcom_push = xcom_push
        self.xcom_key = xcom_key or self.task_id
        self.mcp_client = None
    
    def execute(self, context):
        """
        执行 MCP 工具调用
        
        :param context: Airflow 上下文
        :return: MCP 执行结果
        """
        self.log.info(f"Executing MCP Operator for tool: {self.tool_name}")
        self.log.info(f"Tool params: {self.tool_params}")
        
        # 获取 MCP 连接配置
        from airflow.hooks.base_hook import BaseHook
        conn = BaseHook.get_connection(self.mcp_conn_id)
        mcp_url = f"{conn.schema}://{conn.host}:{conn.port}"
        
        # 创建 MCP 客户端
        self.mcp_client = MCPHTTPClient(
            base_url=mcp_url,
            api_key=conn.password,
            timeout=self.mcp_timeout
        )
        
        # 调用 MCP 工具
        retry_count = 0
        while retry_count <= self.mcp_retry_count:
            try:
                result = self.mcp_client.call_tool(
                    tool_name=self.tool_name,
                    params=self.tool_params,
                    context=context
                )
                self.log.info(f"MCP tool call successful: {self.tool_name}")
                self.log.info(f"Result: {result}")
                
                # 将结果推送到 XCom
                if self.xcom_push:
                    context['ti'].xcom_push(key=self.xcom_key, value=result)
                
                return result
            except MCPException as e:
                retry_count += 1
                if retry_count > self.mcp_retry_count:
                    self.log.error(f"MCP tool call failed after {self.mcp_retry_count} retries: {str(e)}")
                    raise
                self.log.warning(f"MCP tool call failed, retrying in {self.mcp_retry_delay}s: {str(e)}")
                import time
                time.sleep(self.mcp_retry_delay)
            except Exception as e:
                self.log.error(f"Unexpected error during MCP tool call: {str(e)}")
                raise
    
    def on_kill(self):
        """
        任务被终止时的清理操作
        """
        if self.mcp_client:
            self.mcp_client.close()
        self.log.info(f"MCP Operator killed: {self.task_id}")

# 示例：MCP Airflow Operator 使用代码
from airflow import DAG
from airflow.utils.dates import days_ago
from mcp_airflow_operator import MCPOperator
from airflow.operators.dummy_operator import DummyOperator

# 默认参数
default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
    'retries': 1,
    'retry_delay': 300,
}

# 定义 DAG
dag = DAG(
    'mcp_example_dag',
    default_args=default_args,
    description='MCP Airflow 集成示例',
    schedule_interval='@daily',
)

# 开始任务
start = DummyOperator(
    task_id='start',
    dag=dag,
)

# MCP 工具调用任务 1：查询天气
query_weather = MCPOperator(
    task_id='query_weather',
    mcp_conn_id='mcp_default',
    tool_name='weather_api',
    tool_params={
        'city': 'Beijing',
        'date': '{{ ds }}'
    },
    xcom_push=True,
    dag=dag,
)

# MCP 工具调用任务 2：生成旅行建议
generate_travel_plan = MCPOperator(
    task_id='generate_travel_plan',
    mcp_conn_id='mcp_default',
    tool_name='travel_planner',
    tool_params={
        'city': 'Beijing',
        'weather': '{{ ti.xcom_pull(task_ids="query_weather") }}',
        'duration': 3
    },
    xcom_push=True,
    dag=dag,
)

# MCP 工具调用任务 3：发送邮件
send_email = MCPOperator(
    task_id='send_email',
    mcp_conn_id='mcp_default',
    tool_name='email_sender',
    tool_params={
        'to': 'user@example.com',
        'subject': 'Beijing Travel Plan',
        'body': '{{ ti.xcom_pull(task_ids="generate_travel_plan") }}'
    },
    dag=dag,
)

# 结束任务
end = DummyOperator(
    task_id='end',
    dag=dag,
)

# 定义任务依赖
start >> query_weather >> generate_travel_plan >> send_email >> end

# 示例：MCP 动态任务生成代码
from mcp.v2.workflow import DynamicTaskGenerator
from mcp.v2.client import MCPHTTPClient

# 创建 MCP 客户端
mcp_client = MCPHTTPClient(
    base_url="http://localhost:8080",
    api_key="your-api-key"
)

# 创建动态任务生成器
task_generator = DynamicTaskGenerator(
    mcp_client=mcp_client,
    llm_model="gpt-4",
    task_template_path="./task_templates"
)

# 当前工作流状态
current_state = {
    "workflow_id": "wf_001",
    "current_step": 2,
    "execution_history": [
        {
            "task_id": "step1",
            "task_type": "query_weather",
            "status": "completed",
            "result": {"weather": "sunny", "temperature": 20}
        },
        {
            "task_id": "step2",
            "task_type": "analyze_data",
            "status": "completed",
            "result": {"recommendation": "outdoor_activity"}
        }
    ],
    "available_tools": ["weather_api", "activity_planner", "restaurant_finder", "transport_booker"]
}

# 生成动态任务
dynamic_tasks = task_generator.generate_tasks(
    workflow_state=current_state,
    goal="规划一天的户外活动",
    max_tasks=3
)

print("生成的动态任务：")
for task in dynamic_tasks:
    print(f"- 任务 ID: {task['task_id']}")
    print(f"  任务类型: {task['task_type']}")
    print(f"  工具名称: {task['tool_name']}")
    print(f"  参数: {task['params']}")
    print(f"  依赖: {task['dependencies']}")
    print()

生成的动态任务：
- 任务 ID: step3
  任务类型: activity_planner
  工具名称: activity_planner
  参数: {'weather': 'sunny', 'temperature': 20, 'preference': 'outdoor'}
  依赖: ['step2']

- 任务 ID: step4
  任务类型: restaurant_finder
  工具名称: restaurant_finder
  参数: {'activity': '{{ ti.xcom_pull(task_ids="step3") }}', 'preference': 'local_cuisine'}
  依赖: ['step3']

- 任务 ID: step5
  任务类型: transport_booker
  工具名称: transport_booker
  参数: {'activity': '{{ ti.xcom_pull(task_ids="step3") }}', 'restaurant': '{{ ti.xcom_pull(task_ids="step4") }}'}
  依赖: ['step4']

# 示例：MCP 状态同步协议代码
from mcp.v2.state import StateManager, StateSyncProtocol
from mcp.v2.client import MCPHTTPClient

# 创建 MCP 客户端
mcp_client = MCPHTTPClient(
    base_url="http://localhost:8080",
    api_key="your-api-key"
)

# 创建状态管理器
state_manager = StateManager(
    mcp_client=mcp_client,
    storage_backend="redis",
    connection_string="redis://localhost:6379/0"
)

# 创建状态同步协议实例
state_sync = StateSyncProtocol(
    state_manager=state_manager,
    workflow_engine="airflow",
    sync_interval=5,
    conflict_strategy="mcp_priority"
)

# 启动状态同步
state_sync.start()

# 同步工作流任务状态
def sync_workflow_task_state(workflow_id, task_id, state):
    """
    同步工作流任务状态到 MCP
    
    :param workflow_id: 工作流 ID
    :param task_id: 任务 ID
    :param state: 任务状态
    """
    state_data = {
        "workflow_id": workflow_id,
        "task_id": task_id,
        "state": state,
        "timestamp": "2026-01-01T10:00:00Z",
        "metadata": {
            "engine": "airflow",
            "operator": "MCPOperator"
        }
    }
    state_sync.sync_to_mcp(state_data)

# 同步 MCP 执行状态到工作流引擎
def sync_mcp_execution_state(mcp_execution_id, state):
    """
    同步 MCP 执行状态到工作流引擎
    
    :param mcp_execution_id: MCP 执行 ID
    :param state: 执行状态
    """
    state_data = {
        "mcp_execution_id": mcp_execution_id,
        "state": state,
        "timestamp": "2026-01-01T10:00:00Z",
        "metadata": {
            "engine": "mcp",
            "version": "2.0"
        }
    }
    state_sync.sync_to_workflow(state_data)

# 示例：同步工作流任务状态
sync_workflow_task_state("wf_001", "step1", "running")

# 示例：同步 MCP 执行状态
sync_mcp_execution_state("exec_001", "completed")

# 示例：MCP 工作流事件通知代码
from mcp.v2.events import EventManager, EventType
from mcp.v2.client import MCPHTTPClient

# 创建 MCP 客户端
mcp_client = MCPHTTPClient(
    base_url="http://localhost:8080",
    api_key="your-api-key"
)

# 创建事件管理器
event_manager = EventManager(
    mcp_client=mcp_client,
    backend="redis",
    connection_string="redis://localhost:6379/0"
)

# 定义事件处理器
def on_mcp_execution_completed(event):
    """
    MCP 执行完成事件处理器
    
    :param event: 事件对象
    """
    print(f"接收到 MCP 执行完成事件：")
    print(f"- 事件 ID: {event.event_id}")
    print(f"- 执行 ID: {event.execution_id}")
    print(f"- 工具名称: {event.tool_name}")
    print(f"- 执行结果: {event.result}")
    print(f"- 执行时间: {event.timestamp}")
    
    # 触发工作流引擎任务
    trigger_workflow_task(event.execution_id, event.result)

def trigger_workflow_task(execution_id, result):
    """
    触发工作流引擎任务
    
    :param execution_id: MCP 执行 ID
    :param result: 执行结果
    """
    # 这里可以调用工作流引擎 API，触发相应的任务执行
    print(f"触发工作流引擎任务，执行 ID: {execution_id}, 结果: {result}")

# 注册事件处理器
event_manager.register_handler(EventType.MCP_EXECUTION_COMPLETED, on_mcp_execution_completed)

# 启动事件监听
event_manager.start_listening()

# 发布示例事件
sample_event = {
    "event_id": "evt_001",
    "event_type": EventType.MCP_EXECUTION_COMPLETED,
    "execution_id": "exec_001",
    "tool_name": "weather_api",
    "result": {"weather": "sunny", "temperature": 20},
    "timestamp": "2026-01-01T10:00:00Z",
    "metadata": {
        "source": "mcp_server",
        "version": "2.0"
    }
}

event_manager.publish_event(sample_event)

接收到 MCP 执行完成事件：
- 事件 ID: evt_001
- 执行 ID: exec_001
- 工具名称: weather_api
- 执行结果: {'weather': 'sunny', 'temperature': 20}
- 执行时间: 2026-01-01T10:00:00Z
触发工作流引擎任务，执行 ID: exec_001, 结果: {'weather': 'sunny', 'temperature': 20}

# MCP 工作流集成配置文件示例
version: "2.0"

mcp:
  base_url: "http://localhost:8080"
  api_key: "your-api-key"
  timeout: 300
  retry_count: 3
  retry_delay: 10

workflow_engine:
  type: "airflow"
  version: "2.8.0"
  connection_string: "postgresql://airflow:airflow@localhost:5432/airflow"
  webserver_url: "http://localhost:8080"

operator:
  default_timeout: 300
  default_retry_count: 3
  default_retry_delay: 10
  xcom_push: true
  async_execution: true

state_sync:
  enabled: true
  sync_interval: 5
  conflict_strategy: "mcp_priority"
  storage_backend: "redis"
  connection_string: "redis://localhost:6379/0"

event_notification:
  enabled: true
  backend: "redis"
  connection_string: "redis://localhost:6379/0"
  event_types: ["MCP_EXECUTION_STARTED", "MCP_EXECUTION_COMPLETED", "MCP_EXECUTION_FAILED"]

performance:
  connection_pool_size: 100
  cache_enabled: true
  cache_ttl: "1h"
  batch_operations: true
  parallel_execution: true
  max_parallel_tasks: 1000

security:
  authentication: "oauth2"
  authorization: "rbac"
  encryption: "aes-256-gcm"
  audit_logging: true

pip install mcp-airflow-operator

from airflow import DAG
from airflow.utils.dates import days_ago
from mcp_airflow.operator import MCPOperator
from airflow.operators.dummy_operator import DummyOperator

# 默认参数
default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
    'retries': 1,
    'retry_delay': 300,
}

# 定义 DAG
dag = DAG(
    'mcp_example_dag',
    default_args=default_args,
    description='MCP Airflow 集成示例',
    schedule_interval='@daily',
)

# 开始任务
start = DummyOperator(
    task_id='start',
    dag=dag,
)

# MCP 工具调用任务
mcp_task = MCPOperator(
    task_id='mcp_example_task',
    mcp_conn_id='mcp_default',
    tool_name='example_tool',
    tool_params={
        'param1': 'value1',
        'param2': 'value2'
    },
    xcom_push=True,
    dag=dag,
)

# 结束任务
end = DummyOperator(
    task_id='end',
    dag=dag,
)

# 定义任务依赖
start >> mcp_task >> end