Workflows & Sagas

In a complex multi-agent system, an LLM often attempts to perform a sequence of actions. For example, an agent acting as a travel assistant might: 1. Book a flight. 2. Reserve a hotel. 3. Charge the user's credit card.

If step 3 fails, standard MCP frameworks leave the system in a broken state—the flight and hotel are booked, but the user hasn't paid.

AegisMCP prevents this by implementing the Saga Pattern.

Deterministic Sagas

A Saga is a sequence of local transactions. Each transaction has two parts: - execute(): Performs the action. - compensate(): Undoes the action.

If an execute() method fails at any point in the Saga, the WorkflowEngine automatically catches the failure and runs the compensate() methods for all previously successful steps in reverse order.

sequenceDiagram
    participant E as Workflow Engine
    participant S1 as Step 1 (Flight)
    participant S2 as Step 2 (Hotel)
    participant S3 as Step 3 (Credit Card)

    E->>S1: execute()
    S1-->>E: Success
    E->>S2: execute()
    S2-->>E: Success
    E->>S3: execute()
    S3-->>E: Fails (Insufficient Funds)

    rect rgb(60, 20, 20)
        Note over E: Rollback Triggered
        E->>S2: compensate()
        E->>S1: compensate()
    end

State Passing & Compensation

In a real enterprise system, a step needs the result of the previous step to function. Furthermore, the compensate() method needs to know the exact state that was generated during execute() (like a Booking ID) so it knows exactly what to cancel.

AegisMCP handles this securely.

import asyncio
from aegismcp.server.app import AegisMCP
from aegismcp.kernel.context import AegisContext

app = AegisMCP("TravelAgent")

class BookFlightStep:
    async def execute(self, ctx: AegisContext):
        print("Booking flight...")
        booking_id = "FLIGHT_999"

        # We mutate the baggage to pass state to future steps,
        # AND to remember what to compensate!
        ctx = ctx.with_metadata({"flight_id": booking_id})
        return ctx

    async def compensate(self, ctx: AegisContext):
        flight_id = ctx.metadata.get("flight_id")
        print(f"Canceling flight {flight_id}...")

class ChargeCardStep:
    async def execute(self, ctx: AegisContext):
        flight_id = ctx.metadata.get("flight_id")
        print(f"Attempting to charge card for {flight_id}...")
        raise ValueError("Card Declined!")

    async def compensate(self, ctx: AegisContext):
        pass

@app.workflow("book_vacation")
async def book_vacation_saga(ctx: AegisContext):
    # Pass the initial context. The engine automatically threads 
    # the returned, mutated context forward to the next steps!
    return await app.workflow_engine.execute_saga([
        (BookFlightStep(), (), {}),
        (ChargeCardStep(), (), {})
    ], ctx)

Timeout Handling

If an execute() step exceeds the AegisContext.deadline, the ToolExecutor will raise a ToolTimeoutError.

Because a timeout is treated as a failure, the Saga will automatically trigger compensation, ensuring that a hanging API call doesn't leave your system in a stuck, half-booked state!