A Multi-Agent Workflow With MCP Tools in About 60 Lines

A Multi-Agent Workflow With MCP Tools in About 60 Lines

An MCP agent demo is usually one agent calling one tool server. Add a second agent and a hand-off between them, and a boring question shows up fast: if the process dies between agent one and agent two, what runs again?

Below is the smallest version I'd show a team. A researcher agent gets its tools from an MCP server, a writer agent turns the findings into a summary, and a Dapr workflow runs them in order. Each agent call is a workflow activity, so its result is recorded before the next step starts.

What you need

pip install dapr-agents
dapr init

Plus an MCP server you already have running (this example connects over SSE) and an API key for whichever LLM provider you've configured.

The code

import asyncio
import dapr.ext.workflow as wf
from dapr_agents import Agent
from dapr_agents.tool.mcp import MCPClient

wfr = wf.WorkflowRuntime()

async def load_mcp_tools():
    client = MCPClient()
    await client.connect_sse(server_name="research", url="http://localhost:8000/sse")
    return client.get_all_tools()

researcher = Agent(
    name="researcher",
    role="Finds facts with the tools it has",
    instructions=["Use the MCP tools. Return bullet points with sources."],
    tools=asyncio.run(load_mcp_tools()),
)

writer = Agent(
    name="writer",
    role="Turns research notes into a short summary",
    instructions=["Write 150 words. Use only facts from the notes."],
)

@wfr.activity(name="research")
def research(ctx, topic: str) -> str:
    return str(asyncio.run(researcher.run(f"Research: {topic}")))

@wfr.activity(name="write")
def write(ctx, notes: str) -> str:
    return str(asyncio.run(writer.run(f"Summarise these notes:\n{notes}")))

@wfr.workflow(name="research_and_write")
def research_and_write(ctx: wf.DaprWorkflowContext, topic: str):
    notes = yield ctx.call_activity(research, input=topic)
    summary = yield ctx.call_activity(write, input=notes)
    return summary

if __name__ == "__main__":
    wfr.start()
    client = wf.DaprWorkflowClient()
    run_id = "research-demo"
    if client.get_workflow_state(run_id) is None:
        client.schedule_new_workflow(
            workflow=research_and_write,
            input="durable execution for AI agents",
            instance_id=run_id,
        )
    state = client.wait_for_workflow_completion(run_id, timeout_in_seconds=600)
    print(state.serialized_output)
    wfr.shutdown()

Run it with the sidecar:

dapr run --app-id mcp-agents -- python app.py

The fixed run_id matters. On a restart the script finds the existing workflow instead of starting a fresh one, and the runtime carries on with it.

Break it on purpose
Add time.sleep(30) at the top of write (and import time). Start the app, wait until the log shows the research activity completed, then Ctrl+C. Start it again with the same command.

The workflow picks up at write. The researcher's MCP calls don't run a second time, because their result is already in the workflow history. With a plain script you'd be back at the top, calling every MCP tool again and paying for every LLM turn again.

Where this version stops
The checkpoint here is per agent. If the process dies halfway through the researcher, the whole research activity runs again, MCP calls included. That's fine for read-only tools like search. If your MCP agents call tools that write – open an issue, post a message, update a record – swap Agent for DurableAgent, which runs the agent loop itself as a workflow, so each tool call gets its own recorded step.

The researcher and writer are placeholders. Point load_mcp_tools at your own MCP server, run the restart test, and lmk what breaks.