Agentic AI vs MCP

# Orchestrator → Specialist Agent pattern
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent

# Specialist agents (each has focused tools)
researcher = create_react_agent(llm, tools=[web_search, arxiv_search])
writer     = create_react_agent(llm, tools=[draft_doc, format_output])
validator  = create_react_agent(llm, tools=[fact_check, cite_sources])

def router(state):
    # Orchestrator decides next node
    if state["phase"] == "research": return "researcher"
    if state["phase"] == "write":    return "writer"
    if state["phase"] == "validate": return "validator"
    return END

graph = StateGraph(AgentState)
graph.add_node("researcher", researcher)
graph.add_node("writer",     writer)
graph.add_node("validator",  validator)
graph.add_conditional_edges("orchestrator", router)

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

# Connect to MCP servers (tools discovered automatically)
client = MultiServerMCPClient({
    "filesystem": {"command": "npx", "args": ["@modelcontextprotocol/server-filesystem", "/data"]},
    "postgres":   {"url": "http://localhost:8001/sse"},
    "slack":      {"url": "http://localhost:8002/sse"},
})

async def run_agent(task: str):
    # Tools are auto-discovered from MCP servers
    tools = await client.get_tools()

    agent = create_react_agent(
        ChatAnthropic(model="claude-opus-4-5"),
        tools=tools,
        state_modifier="You are a factory planning assistant..."
    )

    async for chunk in agent.astream({"messages": [("user", task)]}):
        print(chunk)  # streamed agent steps

from fastmcp import FastMCP
from pydantic import BaseModel

mcp = FastMCP("factory-tools", description="NeoFAB manufacturing tools")

# ── Tool: callable action ──
@mcp.tool()
def calculate_takt_time(demand_per_day: int, shift_hours: float = 8.0) -> dict:
    """Calculate TAKT time in seconds given daily demand and shift hours."""
    available_seconds = shift_hours * 3600
    takt = available_seconds / demand_per_day
    return {"takt_seconds": round(takt, 2), "demand": demand_per_day}

# ── Resource: read-only data ──
@mcp.resource("machines://catalog")
def get_machine_catalog() -> str:
    """Returns the full machine database as JSON."""
    return load_machine_db().to_json()

# ── Prompt template ──
@mcp.prompt()
def validate_line_prompt(line_name: str, takt: float) -> str:
    return f"Validate machine selections for {line_name} with TAKT={takt}s..."

if __name__ == "__main__":
    mcp.run(transport="stdio")  # or transport="sse" for remote

# Production: run as a service, connect over HTTP
mcp.run(
    transport="sse",
    host="0.0.0.0",
    port=8001,
    # Add auth middleware in production!
)

{
  "mcpServers": {
    "factory-tools": {
      "command": "python",
      "args": ["/path/to/server.py"]
    },
    "remote-db": {
      "url": "http://localhost:8001/sse"
    }
  }
}

from pydantic import BaseModel

class AgentEvalResult(BaseModel):
    correctness:  float   # 0-1: did it get the right answer?
    tool_usage:   float   # 0-1: did it use the right tools?
    efficiency:   float   # 0-1: fewest steps needed?
    faithfulness: float   # 0-1: no hallucinations?
    reasoning:    str     # explanation of scores

# Run judge against agent trace
judge_result = judge_llm.with_structured_output(AgentEvalResult).invoke({
    "task":       original_task,
    "agent_trace": tool_call_history,
    "final_answer": agent_output,
    "ground_truth": expected_answer  # optional
})

# 1. Prompt caching — static prefix cached at ~10% cost
client.messages.create(
    system=[{"type": "text", "text": LARGE_SYSTEM_PROMPT,
             "cache_control": {"type": "ephemeral"}}],  # cache this!
    ...
)

# 2. Model tiering — route by complexity
model = "claude-haiku-4-5" if task.complexity == "low" else "claude-opus-4-5"

# 3. Observation truncation — don't re-inject full tool result
truncated_obs = tool_result[:2000] + "... [truncated]" if len(tool_result) > 2000 else tool_result

# 4. Batch API for offline evals (50% cost savings)
batch = client.beta.messages.batches.create(requests=[...])

def get_station_cycle_time(station_id: str, product_line: str) -> dict:
    """
    Retrieve the measured cycle time for a manufacturing station.

    Use this when you need the actual (measured) cycle time for a station
    to compare against TAKT time. Do NOT use for theoretical times.

    Args:
        station_id: Station identifier (e.g., "ST-003", "WASH-01")
        product_line: Product line code (e.g., "2kWh", "12kWh", "HeroPack")

    Returns:
        {
          "station_id": str,
          "cycle_time_seconds": float,
          "last_measured": ISO8601 date,
          "measurement_count": int
        }

    Errors:
        STATION_NOT_FOUND — station_id does not exist
        LINE_NOT_FOUND    — product_line does not exist
        NO_DATA           — station exists but no measurements yet
    """