MCP (Model Context Protocol) Integration¶

from edda.integrations.mcp import EddaMCPServer
from edda import WorkflowContext, activity

# Create MCP server
server = EddaMCPServer(
    name="Order Service",
    db_url="postgresql://user:pass@localhost/orders",
)

@activity
async def reserve_inventory(ctx: WorkflowContext, items: list[str]):
    # Your business logic here
    return {"reserved": True}

@activity
async def process_payment(ctx: WorkflowContext, amount: float):
    # Payment processing logic
    return {"transaction_id": "txn_123"}

@server.durable_tool(description="Process customer order workflow")
async def process_order(ctx: WorkflowContext, order_id: str, items: list[str]):
    """
    Long-running order processing workflow.

    This workflow reserves inventory, processes payment, and ships the order.
    """
    # Reserve inventory
    await reserve_inventory(ctx, items)

    # Process payment
    await process_payment(ctx, 99.99)

    return {"status": "completed", "order_id": order_id}

# Deploy with uvicorn (production)
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(server.asgi_app(), host="0.0.0.0", port=8000)

from edda.integrations.mcp import EddaMCPServer
from mcp.server.fastmcp.prompts.base import UserMessage
from mcp.types import TextContent

server = EddaMCPServer(
    name="Workflow Analysis Service",
    db_url="sqlite:///workflow.db",
)

@server.prompt(description="Analyze a completed workflow execution")
async def analyze_workflow(instance_id: str) -> UserMessage:
    """Generate an analysis prompt for a workflow."""
    # Access workflow state via server.storage
    instance = await server.storage.get_instance(instance_id)
    history = await server.storage.get_history(instance_id)

    text = f"""**Workflow Analysis Request**

**Workflow Details:**
- Instance ID: {instance_id}
- Status: {instance['status']}
- Workflow: {instance['workflow_name']}
- Activities Completed: {len(history)}

**Execution History:**
{chr(10).join(f"- {h['activity_id']}: {h['event_type']}" for h in history)}

Please analyze this workflow execution and provide:
1. Summary of what happened
2. Any potential issues or improvements
3. Performance observations"""

    return UserMessage(content=TextContent(type="text", text=text))

@server.prompt(description="Debug a failed workflow")
async def debug_workflow(instance_id: str) -> UserMessage:
    """Generate a debugging prompt for failed workflows."""
    instance = await server.storage.get_instance(instance_id)
    history = await server.storage.get_history(instance_id)

    # Find failed activities
    failed = [h for h in history if h.get("event_type") == "ActivityFailed"]

    text = f"""**Workflow Debug Request**

Status: {instance['status']}
Error: {instance.get('error', 'None')}

Failed Activities: {len(failed)}
{chr(10).join(f"- {f['activity_id']}: {f.get('event_data', {}).get('error', 'Unknown')}" for f in failed)}

Please help debug this workflow failure."""

    return UserMessage(content=TextContent(type="text", text=text))

def verify_token(token: str) -> bool:
    # Your token verification logic
    return token == "secret-token-123"

server = EddaMCPServer(
    name="Order Service",
    db_url="postgresql://user:pass@localhost/orders",
    token_verifier=verify_token,
)

server = EddaMCPServer(
    name="Order Service",
    db_url="postgresql://user:pass@localhost/orders",
    outbox_enabled=True,
    broker_url="nats://localhost:4222",
)