Building a Multi-Agent Research Pipeline with MCP (Model Context Protocol): Complete Guide

Introduction to MCP and Multi-Agent Research Pipelines

The Model Context Protocol (MCP) is an open standard that defines how AI models connect to external data sources, tools, and services. Think of it as the "USB-C of AI" — a universal interface that lets any model talk to any tool without bespoke integrations. When combined with a multi-agent architecture, MCP becomes the backbone of a research pipeline where specialized agents collaborate, share context, and produce high-quality outputs.

A multi-agent research pipeline splits a complex research task across several agents, each with a focused role: one plans, one searches, one synthesizes, one critiques. MCP provides the standardized transport layer that lets these agents invoke tools — web search, file retrieval, database queries, code execution — without each agent reinventing the integration glue.

Why This Matters

• Decoupling: Agents and tools evolve independently. Swap a search backend without touching agent logic.
• Reusability: One MCP server can serve many agents and many projects.
• Observability: Standardized tool calls make it easier to log, trace, and debug agent behavior.
• Composability: New capabilities are added by registering new MCP servers, not by rewriting agents.

Architecture Overview

Our pipeline will consist of four agents connected through an MCP client that routes tool calls to one or more MCP servers:

• Planner Agent — decomposes the research question into sub-questions.
• Searcher Agent — uses MCP tools (web search, document retrieval) to gather evidence.
• Synthesizer Agent — combines evidence into a structured report.
• Critic Agent — reviews the report for gaps, contradictions, and citations.

The MCP servers expose tools like web_search, fetch_page, and save_note. The agents never call these tools directly; they emit tool-call intents that the MCP client dispatches.

Prerequisites and Setup

You will need Python 3.10+, an OpenAI-compatible LLM endpoint (or a local model), and the official MCP SDK. Install dependencies in a fresh virtual environment:

python -m venv .venv
source .venv/bin/activate
pip install mcp openai pydantic python-dotenv

Create a .env file with your API keys:

OPENAI_API_KEY=sk-...
SEARCH_API_KEY=...

Building an MCP Server

An MCP server exposes tools through a simple decorator-based API. Below is a minimal server that provides web search and page fetching capabilities. Save it as search_server.py.

from mcp.server.fastmcp import FastMCP
import os, json, httpx

mcp = FastMCP("research-search")

SEARCH_API_KEY = os.getenv("SEARCH_API_KEY")

@mcp.tool()
def web_search(query: str, max_results: int = 5) -> str:
    """Search the web for a query and return titles + URLs."""
    resp = httpx.get(
        "https://api.search.example/v1/search",
        params={"q": query, "count": max_results, "api_key": SEARCH_API_KEY},
        timeout=20,
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])
    return json.dumps([
        {"title": r["title"], "url": r["url"], "snippet": r.get("snippet", "")}
        for r in results
    ])

@mcp.tool()
def fetch_page(url: str) -> str:
    """Fetch and return the main text content of a web page."""
    resp = httpx.get(url, timeout=20, follow_redirects=True)
    resp.raise_for_status()
    # In production, use a proper HTML-to-text parser.
    return resp.text[:8000]

@mcp.tool()
def save_note(note: str, tag: str = "general") -> str:
    """Persist a research note to local storage for later synthesis."""
    os.makedirs("notes", exist_ok=True)
    with open(f"notes/{tag}.md", "a", encoding="utf-8") as f:
        f.write(note + "\n\n---\n\n")
    return f"Saved note under tag '{tag}'."

if __name__ == "__main__":
    mcp.run(transport="stdio")

This server uses the stdio transport, which is the simplest way to run MCP servers as subprocesses of your application. For networked deployments, you can switch to transport="sse" or transport="streamable-http".

Creating the MCP Client

The client launches the server as a subprocess, discovers its tools, and exposes them to the agents. Here is a reusable client wrapper:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio, json

class MCPClient:
    def __init__(self, server_script: str):
        self.server_script = server_script
        self.session = None
        self._tools = []

    async def connect(self):
        params = StdioServerParameters(
            command="python",
            args=[self.server_script],
        )
        self._ctx = stdio_client(params)
        read, write = await self._ctx.__aenter__()
        self.session = ClientSession(read, write)
        await self.session.__aenter__()
        await self.session.initialize()
        result = await self.session.list_tools()
        self._tools = result.tools

    @property
    def tools(self):
        return self._tools

    async def call_tool(self, name: str, arguments: dict) -> str:
        result = await self.session.call_tool(name, arguments)
        # MCP returns content blocks; concatenate text blocks.
        texts = [c.text for c in result.content if hasattr(c, "text")]
        return "\n".join(texts)

    async def close(self):
        await self.session.__aexit__(None, None, None)
        await self._ctx.__aexit__(None, None, None)

Defining the Agent Layer

Each agent is a thin wrapper around an LLM call. The key design choice is that agents receive the tool schemas from the MCP client and emit tool calls that the orchestrator executes. This keeps agents stateless and testable.

from openai import OpenAI
import json

client = OpenAI()

SYSTEM_PROMPTS = {
    "planner": (
        "You are a research planner. Given a research question, break it into "
        "3-6 concrete sub-questions. Return JSON: {\"subquestions\": [...]}."
    ),
    "searcher": (
        "You are a research searcher. For each sub-question, use the web_search "
        "and fetch_page tools to gather evidence. Save key findings with save_note."
    ),
    "synthesizer": (
        "You are a research synthesizer. Read all saved notes and produce a "
        "structured markdown report with sections, citations, and a summary."
    ),
    "critic": (
        "You are a research critic. Review the report for unsupported claims, "
        "missing perspectives, and citation errors. Return JSON: "
        "{\"issues\": [...], \"score\": 1-10}."
    ),
}

def run_agent(role: str, user_message: str, tools=None) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPTS[role]},
        {"role": "user", "content": user_message},
    ]
    kwargs = dict(model="gpt-4o", messages=messages)
    if tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"
    resp = client.chat.completions.create(**kwargs)
    return resp.choices[0].message

Orchestrating the Pipeline

The orchestrator wires agents to the MCP client and drives the pipeline through its stages. It handles the tool-call loop: when an agent returns a tool call, the orchestrator executes it via MCP and feeds the result back.

import asyncio, json, os

def tools_for_openai(mcp_tools):
    return [
        {
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description,
                "parameters": t.inputSchema,
            },
        }
        for t in mcp_tools
    ]

async def run_pipeline(question: str):
    mcp = MCPClient("search_server.py")
    await mcp.connect()
    try:
        tools = tools_for_openai(mcp.tools)

        # Stage 1: Planning
        plan_msg = run_agent("planner", question)
        subquestions = json.loads(plan_msg.content)["subquestions"]
        print(f"[Planner] {len(subquestions)} sub-questions generated")

        # Stage 2: Searching (with tool-call loop)
        search_prompt = "Research these sub-questions:\n" + "\n".join(
            f"- {q}" for q in subquestions
        )
        messages = [
            {"role": "system", "content": SYSTEM_PROMPTS["searcher"]},
            {"role": "user", "content": search_prompt},
        ]

        for _ in range(10):  # cap iterations
            resp = client.chat.completions.create(
                model="gpt-4o", messages=messages, tools=tools, tool_choice="auto"
            )
            msg = resp.choices[0].message
            messages.append(msg)
            if not msg.tool_calls:
                break
            for tc in msg.tool_calls:
                args = json.loads(tc.function.arguments)
                result = await mcp.call_tool(tc.function.name, args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                })
        print("[Searcher] evidence gathering complete")

        # Stage 3: Synthesis
        notes = ""
        for fname in os.listdir("notes"):
            with open(f"notes/{fname}") as f:
                notes += f.read() + "\n"
        synth_msg = run_agent("synthesizer", f"Notes:\n{notes}")
        report = synth_msg.content
        with open("report.md", "w") as f:
            f.write(report)
        print("[Synthesizer] report.md written")

        # Stage 4: Critique
        critic_msg = run_agent("critic", f"Report:\n{report}")
        print(f"[Critic] {critic_msg.content}")

    finally:
        await mcp.close()

if __name__ == "__main__":
    asyncio.run(run_pipeline("What are the trade-offs of retrieval-augmented generation vs. fine-tuning for enterprise LLM deployments?"))

Adding a Second MCP Server

The real power of MCP emerges when you compose multiple servers. Let us add a code execution server so the synthesizer can run quick calculations or generate charts. Save this as code_server.py:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("code-exec")

@mcp.tool()
def run_python(code: str) -> str:
    """Execute Python code in a sandbox and return stdout."""
    import subprocess, tempfile, os
    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
        f.write(code)
        path = f.name
    try:
        out = subprocess.run(
            ["python", path], capture_output=True, text=True, timeout=15
        )
        return out.stdout + out.stderr
    finally:
        os.unlink(path)

if __name__ == "__main__":
    mcp.run(transport="stdio")

To use both servers, launch two MCPClient instances and merge their tool lists. The orchestrator dispatches a tool call to whichever client owns that tool name:

class MultiMCPClient:
    def __init__(self, scripts: list[str]):
        self.clients = [MCPClient(s) for s in scripts]
        self._tool_map = {}

    async def connect(self):
        for c in self.clients:
            await c.connect()
            for t in c.tools:
                self._tool_map[t.name] = c

    @property
    def tools(self):
        merged = []
        for c in self.clients:
            merged.extend(c.tools)
        return merged

    async def call_tool(self, name: str, arguments: dict) -> str:
        owner = self._tool_map.get(name)
        if not owner:
            raise ValueError(f"Unknown tool: {name}")
        return await owner.call_tool(name, arguments)

    async def close(self):
        for c in self.clients:
            await c.close()

Swap MCPClient for MultiMCPClient(["search_server.py", "code_server.py"]) in the orchestrator and the agents instantly gain code execution alongside search.

Best Practices

1. Keep Tools Focused and Composable

Each MCP tool should do one thing well. Instead of a monolithic research_everything tool, expose web_search, fetch_page, and save_note separately. Agents compose them; servers stay simple.

2. Validate Inputs at the Server Boundary

Use Pydantic models or explicit checks inside tool functions. Never trust that the LLM will send well-formed arguments. Return clear error strings rather than raising exceptions, so the agent can recover gracefully.

3. Cap Tool-Call Iterations

Agents can loop indefinitely if a tool keeps returning errors. Always set a maximum iteration count in the orchestrator's tool-call loop, as shown in the example above.

4. Persist Intermediate State

The save_note pattern decouples searching from synthesis. If the pipeline crashes mid-run, you still have the notes. This also makes it easy to re-run synthesis with a different prompt without redoing expensive searches.

5. Log Every Tool Call

Wrap call_tool with structured logging that records the tool name, arguments, latency, and a truncated result. This is invaluable for debugging agent behavior and auditing research provenance.

6. Isolate Untrusted Code Execution

If you expose run_python, run it in a container or sandbox with resource limits. The subprocess approach above is acceptable for local development but is not safe for production workloads with untrusted inputs.

7. Use Streaming Where Possible

For long-running tools, prefer MCP's streaming transport so the agent can display progress. This improves UX in interactive pipelines and prevents timeouts on slow network calls.

Extending the Pipeline

Once the core loop works, common extensions include:

• Parallel searchers: Spawn one searcher per sub-question and merge notes concurrently with asyncio.gather.
• Iterative refinement: Feed the critic's issues back into the synthesizer for a second pass.
• Vector retrieval server: Add an MCP server that exposes semantic_search over a local document corpus.
• Human-in-the-loop: Insert a checkpoint after planning where a human approves or edits the sub-questions before searching begins.

Conclusion

Building a multi-agent research pipeline on top of MCP gives you a clean separation between agent reasoning and tool execution. Agents stay focused on language and judgment; MCP servers handle the messy reality of APIs, files, and sandboxes. By following the architecture in this guide — a planner, searcher, synthesizer, and critic wired through a composable MCP client — you get a pipeline that is easy to debug, straightforward to extend, and resilient to the inevitable changes in both models and data sources. Start with the single-server version, add a second server once the loop is stable, and iterate from there. The protocol is designed to grow with you.