Tool Isolation with MCP: Secure Boundaries for AI Workflows

So far, our orchestrator:

• drafts content
• supervises quality
• pauses for Slack approval
• resumes safely
• writes files

That is structured.

But it is not yet isolated.

Right now, the orchestrator:

• calls Slack directly
• writes files directly
• could theoretically send emails directly

In production systems, this is dangerous.

We want:

The orchestrator decides. Tools execute.

This separation is called tool isolation.

Why Tool Isolation Matters

Without isolation:

• Prompt injection can trigger unintended actions.
• The model can request arbitrary file writes.
• Business logic leaks into LLM prompts.

With isolation:

• Tools have explicit contracts.
• Inputs are validated.
• External actions are controlled.
• The orchestrator cannot “accidentally” do more than allowed.

MCP gives us this clean boundary.

Architecture After MCP

Before:

LangGraph → Slack API
LangGraph → File writes

After:

LangGraph → MCP Server → Slack Tool
                        → File Tool
                        → (Future) Email Tool

LangGraph no longer touches external systems directly.

That is the boundary.

Step 1 — Add a Separate MCP Tool Server

Create a new folder:

ai-editorial-orchestrator/
  mcp_server/
    server.py
    requirements.txt

Step 2 — MCP Server Dependencies

mcp_server/requirements.txt

fastmcp>=2.0
requests>=2.31

Step 3 — Minimal FastMCP Server

mcp_server/server.py

from fastmcp import FastMCP
import os
import requests
from pathlib import Path

mcp = FastMCP("editorial-tools")


@mcp.tool()
def write_markdown_file(filename: str, content: str) -> str:
    """
    Writes a Markdown file to the out/ directory.
    """
    out_dir = Path("../out")
    out_dir.mkdir(exist_ok=True)

    file_path = out_dir / filename
    file_path.write_text(content, encoding="utf-8")

    return f"Wrote {filename}"


@mcp.tool()
def post_slack_message(text: str) -> str:
    """
    Posts a simple Slack message via webhook.
    """
    webhook = os.getenv("SLACK_WEBHOOK_URL")
    if not webhook:
        return "Slack webhook not configured."

    payload = {"text": text}
    requests.post(webhook, json=payload, timeout=10)

    return "Slack message sent."


if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=9000)

Run MCP server:

cd mcp_server
python server.py

Note the explicit transport="http" — FastMCP defaults to stdio (the transport Claude Desktop and similar hosts use to launch a server as a subprocess), so a server meant to be reached over the network, like this one sitting behind Docker Compose, has to opt into HTTP explicitly. With that set, FastMCP exposes a single /mcp endpoint that speaks the Streamable HTTP transport.

Step 4 — Update Docker Compose

Add MCP service:

services:
  orchestrator:
    build: .
    env_file:
      - .env
    depends_on:
      - mcp
    volumes:
      - ./out:/app/out
    ports:
      - "8000:8000"
    command: ["uvicorn", "app.server:app", "--host", "0.0.0.0", "--port", "8000"]

  mcp:
    build: ./mcp_server
    env_file:
      - .env
    ports:
      - "9000:9000"

Now tools run separately.

Step 5 — Call MCP Tools from LangGraph

Inside app/graph.py, replace direct Slack/file calls.

Instead of:

post_slack_message(...)

Call it through the official MCP client rather than hand-rolling REST — MCP speaks JSON-RPC over that /mcp endpoint, with a session handshake in front of it, so a plain requests.post won’t do:

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

MCP_URL = "http://mcp:9000/mcp"


async def _call_mcp_async(tool_name: str, args: dict):
    async with streamablehttp_client(MCP_URL) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool(tool_name, arguments=args)
            return result


def call_mcp(tool_name: str, args: dict):
    """Sync wrapper so this can be called from a regular LangGraph node."""
    return asyncio.run(_call_mcp_async(tool_name, args))

Add mcp>=1.0 to the orchestrator’s requirements.txt for this — it’s the official Python SDK, separate from fastmcp which only the tool server needs.

Example usage:

call_mcp(
    "write_markdown_file",
    {"filename": "newsletter.md", "content": state["newsletter_md"]}
)

And:

call_mcp(
    "post_slack_message",
    {"text": "Newsletter draft approved."}
)

What Changed Conceptually

The orchestrator now:

• cannot access Slack directly
• cannot write arbitrary files
• cannot extend itself silently

It must:

1. Explicitly call a tool
2. Pass structured arguments
3. Respect the tool contract

That is security by design.

Why This Is Powerful

You now have:

• Clear execution boundary
• Isolated side effects
• Replaceable tool layer
• Extendable system (email, Medium, analytics)

And if one day you want:

• Cursor
• Claude Desktop
• Another orchestrator

They can all reuse the same MCP tool server.

What Comes Next

We have completed the architecture layer.

Now the final posts can focus on:

• Production hardening (idempotency + per-run isolation)
• Observability and structured logging
• Final series overview and recap

You’ve moved from:

“Let’s try agents”

to

“Let’s design an orchestration system.”

That’s a big shift.