Agentic AI with Multi-Agent Systems — Part III

LangGraph + MCP: Explicit Orchestration Meets Isolated Enterprise Tools

The same Northstar Cloud incident-control problem, now combining LangGraph workflow control with MCP capability boundaries.

Part III — LangGraph + MCP

Same question. Same Northstar Cloud incident. A third architecture.

Part I: LangGraph + local Python functions
Part II: CrewAI + MCP
Part III: LangGraph + MCP

Part III keeps Part I’s strongest property—explicit state, edges, revision loops, and human escalation—while moving enterprise capabilities behind MCP servers that can be owned by isolated teams and reused by other AI applications.

LangGraph answers: What happens next?
MCP answers: What capability can this node access, and through what contract?

Note

This is a teaching article. Quarto does not execute these examples. Copy the code into the named project files and run it locally.

1. One problem, three architectural answers

Northstar Cloud receives the same incident used in Parts I and II:

Checkout latency and error rate increased shortly after 10:10.

Evidence still comes from telemetry.csv, operations.db, and incident_playbook.pdf. The responsibilities remain:

Triage → Evidence → Response Plan → Risk Critic → Communication / Human

Part I

LangGraph + Python

Graph owns workflow. Application owns tool functions.

Best lesson: state + nodes + routing.

Part II

CrewAI + MCP

CrewAI owns role/task collaboration. MCP servers own capabilities.

Best lesson: roles + tasks + interoperability.

Part III

LangGraph + MCP

LangGraph owns workflow. MCP servers own capabilities.

Best lesson: explicit control + interoperability.

2. Why MCP matters when teams are isolated

Imagine three teams:

Observability Team  → telemetry
Platform Team       → deployments and ownership
SRE Governance Team → approved incident policy

Part I could place all access functions inside the LangGraph application:

Graph node → Python function → data

Part III instead uses:

Graph node → MCP client → team-owned MCP server → data

The graph team owns workflow policy. The service team owns capability implementation.

That is a real architectural boundary—not simply another library.

3. Architecture

🏢 Northstar Incident-Control Architecture

Input🚨 Incident
INC-1042

→

<div class="vbox purple"><span class="arch-label">LangGraph orchestration</span>🧠 Shared State + Routing</div>
<div class="graph-row">
  <div class="graph-node blue">📡 Triage</div>
  <div class="graph-node cyan">🔎 Evidence</div>
  <div class="graph-node green">🧭 Planner</div>
  <div class="graph-node orange">🛡️ Critic</div>
  <div class="graph-node purple">📣 Communication</div>
</div>

→

<div class="vbox blue">📈 Telemetry MCP<br><small>→ telemetry.csv</small></div>
<div class="vbox green">🗄️ Operations MCP<br><small>→ operations.db</small></div>
<div class="vbox orange">📚 Playbook MCP<br><small>→ incident_playbook.pdf</small></div>

LangGraph decides the path; MCP exposes the capabilities.

Two responsibilities must remain separate:

\boxed{\text{LangGraph = workflow control}} \qquad \boxed{\text{MCP = capability boundary}}

4. What you design vs. what libraries handle

Concern	You design	LangGraph/MCP handle
business objective	✓
state fields	✓	state transport
node responsibilities	✓	node execution
routing/escalation	✓	edge execution
tool meaning	✓	MCP schema/protocol
permissions	✓	enforcement mechanisms
model prompts	✓	model invocation integration
transport details	configure	MCP client/server mechanics

Neither framework nor protocol knows your organization’s definition of a safe incident response.

5. Project structure

northstar_langgraph_mcp/
├── app.py
├── config.json
├── requirements.txt
├── data/
│   ├── telemetry.csv
│   ├── operations.db
│   └── incident_playbook.pdf
├── servers/
│   ├── telemetry_server.py
│   ├── operations_server.py
│   └── playbook_server.py
├── graph/
│   ├── state.py
│   ├── mcp_gateway.py
│   ├── nodes.py
│   ├── routing.py
│   └── workflow.py
└── tests/
    ├── test_mcp.py
    └── test_routing.py

servers/ are capability providers. graph/ is the orchestration consumer.

6. Install

🧰 What this command does
Creates an isolated Python 3.11 environment and installs the orchestration, LLM, data, PDF, and MCP dependencies used by the files that follow.

➡️ Your action: Run this once from the project root.

python3.11 -m venv .venv
source .venv/bin/activate
pip install langgraph langchain-openai pydantic pandas pypdf "mcp[cli]"

Create config.json locally. This is runtime configuration, not article content; it supplies the model and secret only when you run the project.

➡️ Your action: Create this file locally and never commit the real key.

{
  "OPENAI_API_KEY": "your-key-here",
  "MODEL": "gpt-4o-mini"
}

And .gitignore:

config.json
.venv/
__pycache__/

7. MCP server 1 — telemetry

Create servers/telemetry_server.py:

🧩 What this code is — Telemetry MCP server
This file belongs to the Observability capability layer. It reads telemetry and exposes only two narrow, read-only operations. The next step is the Operations server; later, the Evidence node will call both through the MCP gateway.

➡️ Your action: Create servers/telemetry_server.py. Keep storage details here—not in LangGraph nodes.

💻 View / copy code

from pathlib import Path
import json
import pandas as pd
from mcp.server import MCPServer

ROOT = Path(__file__).resolve().parents[1]
CSV_PATH = ROOT / "data" / "telemetry.csv"

mcp = MCPServer("northstar-telemetry",
                instructions="Read-only Northstar telemetry.")

def _load():
    return pd.read_csv(CSV_PATH)

@mcp.tool()
def get_service_metrics(service: str, last_n: int = 6) -> str:
    """Return recent telemetry for one service."""
    df = _load()
    rows = df[df["service"] == service].tail(last_n)
    return json.dumps(rows.to_dict(orient="records"))

@mcp.tool()
def summarize_service_metrics(service: str) -> str:
    """Return a compact telemetry summary."""
    x = _load()
    x = x[x["service"] == service]
    if x.empty:
        return json.dumps({"error": "service not found"})
    return json.dumps({
        "service": service,
        "latest_latency_ms": float(x.iloc[-1]["p95_latency_ms"]),
        "latest_error_rate": float(x.iloc[-1]["error_rate"]),
        "max_latency_ms": float(x["p95_latency_ms"].max()),
        "max_error_rate": float(x["error_rate"].max()),
    })

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

The owning team can later replace CSV with an observability API without changing the public tool contract.

8. MCP server 2 — operations

Create servers/operations_server.py:

🧩 What this code is — Operations MCP server
This server isolates deployment and customer-impact access. The graph never receives arbitrary SQL access; it receives business-shaped functions such as get_recent_deployments().

➡️ Your action: Create servers/operations_server.py. If the database changes later, preserve these contracts where possible.

💻 View / copy code

from pathlib import Path
import json
import sqlite3
from mcp.server import MCPServer

ROOT = Path(__file__).resolve().parents[1]
DB_PATH = ROOT / "data" / "operations.db"

mcp = MCPServer("northstar-operations",
                instructions="Read-only operational metadata.")

def _query(sql: str, params: tuple):
    with sqlite3.connect(DB_PATH) as conn:
        conn.row_factory = sqlite3.Row
        rows = conn.execute(sql, params).fetchall()
    return [dict(r) for r in rows]

@mcp.tool()
def get_recent_deployments(service: str, limit: int = 5) -> str:
    """Return recent deployments."""
    rows = _query(
        """SELECT service, version, deployed_at, status
           FROM deployments WHERE service = ?
           ORDER BY deployed_at DESC LIMIT ?""",
        (service, limit),
    )
    return json.dumps(rows)

@mcp.tool()
def get_customer_impact(service: str) -> str:
    """Return recent customer-impact records."""
    rows = _query(
        """SELECT service, impacted_customers, region, observed_at
           FROM customer_impact WHERE service = ?
           ORDER BY observed_at DESC LIMIT 5""",
        (service,),
    )
    return json.dumps(rows)

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

Notice what we did not expose: execute_arbitrary_sql(sql). Narrow capabilities are easier to secure and audit.

9. MCP server 3 — playbook

Create servers/playbook_server.py:

🧩 What this code is — Playbook MCP server
This is the policy/retrieval boundary. It turns the PDF into a searchable capability. The simple keyword retrieval is intentionally replaceable by vector/RAG retrieval without changing the graph.

➡️ Your action: Create servers/playbook_server.py, then test all three MCP servers before building agents.

💻 View / copy code

from pathlib import Path
import re
from pypdf import PdfReader
from mcp.server import MCPServer

ROOT = Path(__file__).resolve().parents[1]
PDF_PATH = ROOT / "data" / "incident_playbook.pdf"

mcp = MCPServer("northstar-playbook",
                instructions="Approved incident-response guidance.")

def _terms(text: str):
    return {w.lower() for w in re.findall(r"[A-Za-z]{4,}", text)}

@mcp.tool()
def search_playbook(query: str, top_k: int = 3) -> str:
    """Return relevant playbook passages."""
    q = _terms(query)
    scored = []
    for page_no, page in enumerate(PdfReader(PDF_PATH).pages, start=1):
        text = page.extract_text() or ""
        score = len(q & _terms(text))
        if score:
            scored.append((score, page_no, text))
    scored.sort(reverse=True)
    parts = []
    for score, page_no, text in scored[:top_k]:
        parts.append(
            f"PAGE {page_no} | score={score}\n"
            + " ".join(text.split())[:1800]
        )
    return "\n\n".join(parts) or "No relevant guidance found."

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

For production, the retrieval implementation can become vector/RAG search while the MCP function name and inputs remain stable.

10. Test MCP before LangGraph

Run each server with the MCP Inspector. This validates the capability layer before LangGraph is involved.

➡️ Your action: Confirm that each tool is discoverable and returns sensible data. Only then continue to the graph.

mcp dev servers/telemetry_server.py
mcp dev servers/operations_server.py
mcp dev servers/playbook_server.py

Data

→

MCP Tool

→

MCP Test

→

Graph Node

→

Routing

Debug one layer at a time.

11. Create the MCP gateway

Create graph/mcp_gateway.py:

🧩 What this code is — MCP gateway
The three servers now exist, but LangGraph should not repeat connection code in every node. This gateway is the adapter: nodes ask for a server, tool, and arguments; the gateway handles the MCP client call.

➡️ Your action: Create graph/mcp_gateway.py. All later graph nodes call this one adapter.

💻 View / copy code

import sys
from pathlib import Path
from mcp import Client, StdioServerParameters

ROOT = Path(__file__).resolve().parents[1]

def _server(filename: str):
    return StdioServerParameters(
        command=sys.executable,
        args=[str(ROOT / "servers" / filename)],
    )

SERVERS = {
    "telemetry": _server("telemetry_server.py"),
    "operations": _server("operations_server.py"),
    "playbook": _server("playbook_server.py"),
}

async def call_mcp_tool(server_name: str,
                        tool_name: str,
                        arguments: dict) -> str:
    async with Client(SERVERS[server_name]) as client:
        result = await client.call_tool(tool_name, arguments)
        pieces = []
        for item in result.content:
            value = getattr(item, "text", None)
            if value:
                pieces.append(value)
        return "\n".join(pieces)

Graph nodes now need to know only:

🧩 What this code is — One example MCP call
This tiny example shows the contract that graph code will use. Notice that the caller knows the capability name and inputs, but not SQLite, CSV parsing, or server internals.

➡️ Your action: Use this pattern inside nodes; do not scatter raw database/file access through the graph.

💻 View / copy code

async def example_call():
    return await call_mcp_tool(
        "operations",
        "get_recent_deployments",
        {"service": "checkout-api"},
    )

12. Define outputs and shared state

Create graph/state.py:

🧩 What this code is — State and structured outputs
Before writing nodes, define the shared case file and the model outputs. This prevents each agent from inventing a different data shape and makes routing fields such as decision predictable.

➡️ Your action: Create graph/state.py. The next section’s nodes will read and update these fields.

💻 View / copy code

from typing import Literal, TypedDict
from pydantic import BaseModel, Field

class TriageOutput(BaseModel):
    severity: Literal["LOW", "MEDIUM", "HIGH", "CRITICAL"]
    affected_service: str
    summary: str
    confidence: float = Field(ge=0, le=1)

class PlanOutput(BaseModel):
    action: str
    justification: str
    risk: Literal["LOW", "MEDIUM", "HIGH"]
    requires_human_approval: bool

class CriticOutput(BaseModel):
    decision: Literal["ACCEPT", "REVISE", "ESCALATE"]
    reason: str

class IncidentState(TypedDict, total=False):
    incident: dict
    triage: dict
    evidence: dict
    plan: dict
    critique: dict
    stakeholder_message: str
    revision_count: int
    needs_human: bool
    trajectory: list[str]

This explicit shared case file is one of LangGraph’s strongest teaching features.

13. Build the nodes

Create graph/nodes.py:

🧩 What this code is — LangGraph nodes
These functions implement the five responsibilities. Triage interprets telemetry; Evidence gathers facts deterministically; Planner proposes; Critic reviews; Communication writes the approved message. Each node returns only the state updates it owns.

➡️ Your action: Create graph/nodes.py. Notice how MCP calls connect the orchestration layer to the servers created earlier.

💻 View / copy code

import json
from langchain_core.messages import HumanMessage, SystemMessage

from graph.mcp_gateway import call_mcp_tool
from graph.state import CriticOutput, IncidentState, PlanOutput, TriageOutput

async def triage_node(state: IncidentState, llm):
    service = state["incident"]["service"]
    metrics = await call_mcp_tool(
        "telemetry", "summarize_service_metrics", {"service": service}
    )
    model = llm.with_structured_output(TriageOutput)
    result = await model.ainvoke([
        SystemMessage(content=(
            "You are Signal Triage. Classify severity from supplied "
            "evidence. Do not diagnose root cause."
        )),
        HumanMessage(content=f"Incident: {state['incident']}\nMetrics: {metrics}")
    ])
    return {
        "triage": result.model_dump(),
        "trajectory": state.get("trajectory", []) + ["triage"],
    }

async def evidence_node(state: IncidentState, llm):
    service = state["incident"]["service"]
    metrics = await call_mcp_tool(
        "telemetry", "get_service_metrics",
        {"service": service, "last_n": 6}
    )
    deployments = await call_mcp_tool(
        "operations", "get_recent_deployments",
        {"service": service, "limit": 5}
    )
    impact = await call_mcp_tool(
        "operations", "get_customer_impact", {"service": service}
    )
    policy = await call_mcp_tool(
        "playbook", "search_playbook",
        {"query": f"{service} latency errors deployment"}
    )
    return {
        "evidence": {
            "metrics": metrics,
            "deployments": deployments,
            "customer_impact": impact,
            "playbook": policy,
        },
        "trajectory": state.get("trajectory", []) + ["evidence"],
    }

async def planner_node(state: IncidentState, llm):
    model = llm.with_structured_output(PlanOutput)
    result = await model.ainvoke([
        SystemMessage(content=(
            "You are the Response Planner. Recommend the safest next "
            "action grounded in evidence and policy. Correlation is not causation."
        )),
        HumanMessage(content=(
            f"Evidence: {json.dumps(state['evidence'])}\n"
            f"Critic feedback: {json.dumps(state.get('critique', {}))}"
        )),
    ])
    return {
        "plan": result.model_dump(),
        "trajectory": state.get("trajectory", []) + ["planner"],
    }

async def critic_node(state: IncidentState, llm):
    policy = await call_mcp_tool(
        "playbook", "search_playbook",
        {"query": f"approval escalation {state['plan']['action']}"}
    )
    model = llm.with_structured_output(CriticOutput)
    result = await model.ainvoke([
        SystemMessage(content=(
            "You are the Risk Critic. Check evidence support, policy, "
            "uncertainty and operational risk. Return ACCEPT, REVISE or ESCALATE."
        )),
        HumanMessage(content=(
            f"Evidence: {json.dumps(state['evidence'])}\n"
            f"Plan: {json.dumps(state['plan'])}\nPolicy: {policy}"
        )),
    ])
    revisions = state.get("revision_count", 0)
    if result.decision == "REVISE":
        revisions += 1
    return {
        "critique": result.model_dump(),
        "revision_count": revisions,
        "trajectory": state.get("trajectory", [])
                      + [f"critic:{result.decision}"],
    }

async def communication_node(state: IncidentState, llm):
    response = await llm.ainvoke([
        SystemMessage(content=(
            "Write a concise stakeholder update using reviewed facts only. "
            "Do not expose secrets or present hypotheses as facts."
        )),
        HumanMessage(content=(
            f"Incident: {state['incident']}\n"
            f"Plan: {state['plan']}\nCritic: {state['critique']}"
        )),
    ])
    return {
        "stakeholder_message": response.content,
        "trajectory": state.get("trajectory", []) + ["communication"],
    }

async def human_escalation_node(state: IncidentState):
    return {
        "needs_human": True,
        "trajectory": state.get("trajectory", []) + ["human_escalation"],
    }

The key division is:

LangGraph node → WHEN this work happens
MCP server     → HOW enterprise data is accessed
LLM            → WHERE interpretation is useful
Python rule    → WHAT must be deterministic

14. Make safety routing deterministic

Create graph/routing.py:

🧩 What this code is — Deterministic routing
The Critic produces a judgment, but it does not get unrestricted control of the workflow. This pure Python function converts the judgment into an allowed next step and caps revision loops.

➡️ Your action: Create graph/routing.py. This is the safety/control bridge between the Critic and the graph.

💻 View / copy code

MAX_REVISIONS = 1

def route_after_critic(state):
    decision = state["critique"]["decision"]

    if decision == "ACCEPT":
        if state["plan"]["requires_human_approval"]:
            return "human_escalation"
        return "communication"

    if decision == "REVISE":
        if state.get("revision_count", 0) <= MAX_REVISIONS:
            return "planner"
        return "human_escalation"

    return "human_escalation"

The LLM can recommend. Python owns the hard safety boundary.

15. Build the graph

Create graph/workflow.py:

🧩 What this code is — Workflow construction
Now we connect previously defined pieces: state → nodes → fixed edges → conditional edge after the Critic. The graph contains no database logic; it only describes execution order.

➡️ Your action: Create graph/workflow.py. This is where the architecture becomes an executable state machine.

💻 View / copy code

from langgraph.graph import END, START, StateGraph
from graph.routing import route_after_critic
from graph.state import IncidentState

def build_workflow(triage, evidence, planner, critic,
                   communication, human_escalation):
    graph = StateGraph(IncidentState)

    graph.add_node("triage", triage)
    graph.add_node("evidence", evidence)
    graph.add_node("planner", planner)
    graph.add_node("critic", critic)
    graph.add_node("communication", communication)
    graph.add_node("human_escalation", human_escalation)

    graph.add_edge(START, "triage")
    graph.add_edge("triage", "evidence")
    graph.add_edge("evidence", "planner")
    graph.add_edge("planner", "critic")

    graph.add_conditional_edges(
        "critic",
        route_after_critic,
        {
            "planner": "planner",
            "communication": "communication",
            "human_escalation": "human_escalation",
        },
    )

    graph.add_edge("communication", END)
    graph.add_edge("human_escalation", END)
    return graph.compile()

🔁 LangGraph Decision Loop

🚦 START

→

📡 Triage

→

🔎 Evidence

→

🧭 Planner

→

🛡️ Critic
Decision?

✅ ACCEPT
Low-risk approved plan
↓
📣 Communication → END

🔄 REVISE
Retry still available
↓
🧭 Back to Planner

🧑‍💼 ESCALATE
High risk / retry exhausted
↓
Human Review → END

16. Complete `app.py`

🧩 What this code is — Application assembly
This is the composition root. It loads configuration, creates one LLM, injects it into node wrappers, compiles the workflow, creates the initial incident state, and invokes the graph.

➡️ Your action: Create app.py. At this point all earlier files finally connect end to end.

💻 View / copy code

import asyncio
import json
import os
from pathlib import Path
from langchain_openai import ChatOpenAI

from graph.nodes import (
    triage_node, evidence_node, planner_node, critic_node,
    communication_node, human_escalation_node,
)
from graph.workflow import build_workflow

ROOT = Path(__file__).resolve().parent
with open(ROOT / "config.json", "r", encoding="utf-8") as f:
    config = json.load(f)

os.environ["OPENAI_API_KEY"] = config["OPENAI_API_KEY"]
llm = ChatOpenAI(
    model=config.get("MODEL", "gpt-4o-mini"),
    temperature=0,
)

async def triage(s): return await triage_node(s, llm)
async def evidence(s): return await evidence_node(s, llm)
async def planner(s): return await planner_node(s, llm)
async def critic(s): return await critic_node(s, llm)
async def communication(s): return await communication_node(s, llm)

app = build_workflow(
    triage, evidence, planner, critic,
    communication, human_escalation_node,
)

async def main():
    initial_state = {
        "incident": {
            "incident_id": "INC-1042",
            "service": "checkout-api",
            "description": "Latency and errors increased shortly after 10:10.",
        },
        "revision_count": 0,
        "needs_human": False,
        "trajectory": [],
    }

    result = await app.ainvoke(initial_state)

    print("Trajectory:")
    for step in result["trajectory"]:
        print(" -", step)

    print("Human review:", result.get("needs_human", False))
    if result.get("stakeholder_message"):
        print(result["stakeholder_message"])

if __name__ == "__main__":
    asyncio.run(main())

Run:

python app.py

17. Follow one incident

INC-1042
   ↓
TRIAGE ───────────→ Telemetry MCP
   ↓
EVIDENCE ─────────→ Telemetry + Operations + Playbook MCP
   ↓
PLANNER
   ↓
CRITIC ───────────→ Playbook MCP
   ├─ ACCEPT ─────→ Communication → END
   ├─ REVISE ─────→ Planner
   └─ ESCALATE ───→ Human → END

The workflow is dynamic, but the safety rules remain visible.

18. Observation is not hypothesis is not decision

OBSERVATION:
Latency rose from about 220 ms to about 980 ms.

OBSERVATION:
Version 4.8 was deployed four minutes earlier.

HYPOTHESIS:
The deployment may be related.

DECISION:
Pause rollout and request rollback approval.

Never silently transform “after” into “because of.”

\text{Evidence} \rightarrow \text{Hypothesis} \rightarrow \text{Reviewed Decision}

19. Why MCP helps isolated teams

🤝 Isolated Teams, Stable MCP Contracts

Consumer🧠 LangGraph Team
owns workflow

→

<div class="vbox blue">📡 Observability Team<br><small>Telemetry MCP</small></div>
<div class="vbox green">🧰 Platform Team<br><small>Operations MCP</small></div>
<div class="vbox orange">🛡️ Governance Team<br><small>Playbook MCP</small></div>

→

<div class="vbox blue">📈 Telemetry Platform</div>
<div class="vbox green">🗄️ Deployment DB</div>
<div class="vbox orange">📚 Approved Policy</div>

The graph consumes contracts. It need not own every storage implementation.

20. Local today, remote tomorrow

For learning:

LangGraph → stdio → local MCP subprocess

For deployment:

LangGraph service → authenticated HTTPS → remote MCP service

The current MCP client can connect to a URL:

🧩 What this code is — Remote MCP client
Local stdio is ideal for learning. This snippet shows the architectural evolution: the graph can consume a remote MCP endpoint while keeping the same business workflow.

➡️ Your action: Use remote transport only after adding authentication, authorization, TLS, timeouts, and operational monitoring.

💻 View / copy code

from mcp import Client

async def list_remote_tools():
    async with Client("https://mcp.example.com/mcp") as client:
        return await client.list_tools()

Transport changes without changing the business workflow.

21. MCP is not automatic security

Prefer:

get_recent_deployments(service)
search_playbook(query)
get_customer_impact(service)

over:

execute_any_sql(sql)
read_any_file(path)
run_shell(command)

Production systems still need authentication, authorization, TLS, read/write separation, audit logs, rate limits, timeouts, input validation, and human approval for consequential actions.

Protocol compatibility is not permission.

22. Handle MCP failure explicitly

Do not convert:

MCP unavailable

into:

No evidence found

Those are different states.

Extend state:

🧩 What this code is — Tool-health state
A missing server is not the same as ‘no evidence’. Adding explicit tool errors to state lets the graph preserve that distinction.

➡️ Your action: Add this field to the production state model if external MCP availability affects safe decisions.

💻 View / copy code

class ToolHealthState(TypedDict, total=False):
    tool_errors: list[str]

Then route safely:

🧩 What this code is — Tool-health route
This route turns infrastructure uncertainty into a safe workflow action. If required evidence cannot be fetched, the graph escalates instead of letting the LLM guess.

➡️ Your action: Place this routing rule before planning in a production graph.

💻 View / copy code

def evidence_health_route(state):
    if state.get("tool_errors"):
        return "human_escalation"
    return "planner"

This is where LangGraph’s explicit routing becomes especially valuable.

23. Test MCP without the LLM

Create tests/test_mcp.py:

🧩 What this code is — MCP smoke test
This test deliberately excludes LangGraph and the LLM. It answers one question only: can the MCP client discover and call the telemetry capability?

➡️ Your action: Create tests/test_mcp.py. Run it before debugging agent reasoning.

💻 View / copy code

import anyio
import sys
from pathlib import Path
from mcp import Client, StdioServerParameters

ROOT = Path(__file__).resolve().parents[1]

server = StdioServerParameters(
    command=sys.executable,
    args=[str(ROOT / "servers" / "telemetry_server.py")],
)

async def main():
    async with Client(server) as client:
        tools = await client.list_tools()
        names = [tool.name for tool in tools.tools]
        assert "summarize_service_metrics" in names

        result = await client.call_tool(
            "summarize_service_metrics",
            {"service": "checkout-api"},
        )
        assert result.content
        print("MCP smoke test passed.")

if __name__ == "__main__":
    anyio.run(main())

24. Test routing without the LLM

Create tests/test_routing.py:

🧩 What this code is — Routing unit tests
These tests deliberately exclude MCP and the LLM. They prove that ACCEPT, approval-required, and ESCALATE decisions go to the expected destinations.

➡️ Your action: Create tests/test_routing.py. Hard workflow policy should be cheap and deterministic to test.

💻 View / copy code

from graph.routing import route_after_critic

def test_accept():
    state = {
        "critique": {"decision": "ACCEPT"},
        "plan": {"requires_human_approval": False},
        "revision_count": 0,
    }
    assert route_after_critic(state) == "communication"

def test_approval():
    state = {
        "critique": {"decision": "ACCEPT"},
        "plan": {"requires_human_approval": True},
        "revision_count": 0,
    }
    assert route_after_critic(state) == "human_escalation"

def test_escalation():
    state = {
        "critique": {"decision": "ESCALATE"},
        "plan": {"requires_human_approval": False},
        "revision_count": 0,
    }
    assert route_after_critic(state) == "human_escalation"

Critical routing can therefore be tested without tokens or stochastic model output.

25. Evaluate the system

Use a compact test matrix:

Case	Situation	Expected behavior
A	strong consistent evidence	communicate reviewed plan
B	ambiguous cause	revise or human
C	high-impact action	human approval
D	MCP outage	fail safely / human
E	prompt injection in incident text	ignore malicious instruction

Measure task success, routing correctness, MCP tool correctness, evidence grounding, escalation correctness, unsupported claims, and latency.

26. Part I → Part II → Part III

🗺️ Part I

LangGraph + Python

Learn explicit state, nodes, edges, routing, and deterministic control.

Application owns tool functions.

→

🚪 Part II

CrewAI + MCP

Learn role/task collaboration and standardized capability access.

CrewAI organizes agents; MCP separates tools.

→

III

🏗️ Part III

LangGraph + MCP

Combine explicit workflow control with independently owned enterprise capabilities.

Control + interoperability.

The series now answers three progressively different questions:

Part I:
How do I explicitly control a multi-agent workflow?

Part II:
How do role-based agents consume independently exposed capabilities?

Part III:
Can I retain explicit graph control AND use independent capability boundaries?

Answer: Yes.

27. Three-way comparison

Dimension	LangGraph + Python	CrewAI + MCP	LangGraph + MCP
explicit state	★★★★★	★★★	★★★★★
explicit branching	★★★★★	★★★	★★★★★
role/task simplicity	★★★	★★★★★	★★★
prototype simplicity	★★★★★	★★★	★★★
capability reuse	★★	★★★★★	★★★★★
isolated-team ownership	★★	★★★★★	★★★★★
deterministic safety routing	★★★★★	★★★★	★★★★★
integration boundaries	★★	★★★★★	★★★★★
moving parts	low	medium/high	high
enterprise modularity	medium	high	high

These stars are design guidance for this scenario, not benchmark results.

28. Which architecture should you choose?

Choose LangGraph + Python when one team owns the application and integrations, tools are application-specific, deployment simplicity matters, and explicit routing is important.

Choose CrewAI + MCP when role/task modeling matches the organization, rapid multi-agent collaboration matters, and reusable remote capabilities are important.

Choose LangGraph + MCP when explicit state, branching, retries, and human approval matter and capabilities belong to isolated teams or need reuse across AI applications.

29. Verdict for Northstar Cloud

For the first prototype, I would start with:

\boxed{\text{LangGraph + Python}}

It teaches the mechanics with the fewest moving pieces.

For a role-oriented agent team, Part II is elegant:

\boxed{\text{CrewAI + MCP}}

For the mature Northstar incident platform with independently owned observability, operations, and governance systems, I would lean toward:

\boxed{\text{LangGraph + MCP}}

The reason is structural:

WHAT HAPPENS NEXT?          → LangGraph
WHAT CAN THIS NODE ACCESS?  → MCP
WHAT NEEDS JUDGMENT?        → LLM
WHAT MUST BE GUARANTEED?    → deterministic code + human

30. No universal winner

Part III also costs more. It adds server processes, client lifecycle, protocol boundaries, remote authentication, additional observability, and more failure modes.

If one small team owns everything, MCP can be unnecessary ceremony.

A useful rule is:

\boxed{\text{Use MCP when the capability boundary is real—not merely fashionable.}}

31. Final architecture lesson

The three articles reveal that there are actually three independent design questions:

How should reasoning be decomposed?
Triage → Evidence → Plan → Critic → Communication.
How should workflow be orchestrated?
LangGraph graph/state/edges or CrewAI crew/tasks/flows.
How should external capabilities be exposed?
Local Python functions or MCP servers.

Therefore the design space is larger than “LangGraph versus CrewAI”:

LangGraph + Python
LangGraph + MCP
CrewAI + Python
CrewAI + MCP
other orchestrator + MCP

MCP does not replace orchestration.
LangGraph does not replace capability ownership.
CrewAI does not replace security policy.
An LLM does not replace deterministic engineering.

32. Closing perspective

Part I gave the agents a map.

Part II gave agents standard doors into independently owned systems.

Part III keeps the map and installs those doors.

Business Goal

→

LangGraph State

→

Specialist Nodes

→

MCP Contracts

→

Isolated Teams

The final mental model is:

\boxed{ \text{LangGraph = orchestration} + \text{MCP = capability boundary} + \text{LLM = bounded judgment} + \text{Python/Human = control} }

That is the architectural conclusion of the three-part series.

Part III — LangGraph + MCP

1. One problem, three architectural answers

Part I

Part II

Part III

2. Why MCP matters when teams are isolated

3. Architecture

4. What you design vs. what libraries handle

5. Project structure

6. Install

7. MCP server 1 — telemetry

8. MCP server 2 — operations

9. MCP server 3 — playbook

10. Test MCP before LangGraph

11. Create the MCP gateway

12. Define outputs and shared state

13. Build the nodes

14. Make safety routing deterministic

15. Build the graph

16. Complete app.py

17. Follow one incident

18. Observation is not hypothesis is not decision

19. Why MCP helps isolated teams

20. Local today, remote tomorrow

21. MCP is not automatic security

22. Handle MCP failure explicitly

23. Test MCP without the LLM

24. Test routing without the LLM

25. Evaluate the system

26. Part I → Part II → Part III

🗺️ Part I

🚪 Part II

🏗️ Part III

27. Three-way comparison

28. Which architecture should you choose?

29. Verdict for Northstar Cloud

30. No universal winner

31. Final architecture lesson

32. Closing perspective

16. Complete `app.py`