MCP for Agentic Systems: The Missing Protocol for Autonomous AI

# mcp_models.py
from pydantic import BaseModel, Field
from typing import Any, List, Dict, Literal

class ToolCall(BaseModel):
    tool_name: str
    arguments: Dict[str, Any]

class ActionHistoryItem(BaseModel):
    action_type: Literal["TOOL_CALL", "LLM_RESPONSE", "USER_INPUT"]
    content: ToolCall | str
    result: str | None = None

class AgentState(BaseModel):
    session_id: str

    # Core Agent Components
    objective: str
    plan: List[str] = Field(default_factory=list)
    scratchpad: List[str] = Field(default_factory=list)
    action_history: List[ActionHistoryItem] = Field(default_factory=list)
    world_model: Dict[str, Any] = Field(default_factory=dict)
    
    # Status
    status: Literal["RUNNING", "PLANNING", "WAITING_FOR_TOOL", "SUCCESS", "FAILED"] = "RUNNING"

# MCP message formats
class MCPRequest(BaseModel):
    action: Literal["REQUEST_LLM_COMPLETION", "REQUEST_TOOL_EXECUTION", "FINALIZE_OBJECTIVE"]
    payload: Dict[str, Any]

class MCPResponse(BaseModel):
    success: bool
    data: Dict[str, Any] | None = None
    error: str | None = None

# mcp_gateway.py
import uuid
from fastapi import FastAPI, HTTPException
from mcp_models import AgentState, MCPRequest, MCPResponse, ActionHistoryItem, ToolCall
import your_llm_library # Placeholder for your LLM client (e.g., OpenAI, Anthropic)
import available_tools   # Placeholder for a module defining your agent's tools

app = FastAPI()

# WARNING: In-memory store. Replace with Redis/Postgres for production.
STATE_STORE: Dict[str, AgentState] = {}

@app.post("/sessions", response_model=AgentState)
async def create_session(objective: str):
    """Initializes a new agent session."""
    session_id = str(uuid.uuid4())
    initial_state = AgentState(session_id=session_id, objective=objective)
    STATE_STORE[session_id] = initial_state
    return initial_state

@app.get("/sessions/{session_id}", response_model=AgentState)
async def get_session_state(session_id: str):
    if session_id not in STATE_STORE:
        raise HTTPException(status_code=404, detail="Session not found")
    return STATE_STORE[session_id]

@app.post("/sessions/{session_id}/execute", response_model=MCPResponse)
async def execute_mcp_action(session_id: str, request: MCPRequest):
    """The main entrypoint for the agent's Think-Act loop."""
    if session_id not in STATE_STORE:
        raise HTTPException(status_code=404, detail="Session not found")
    
    state = STATE_STORE[session_id]

    match request.action:
        case "REQUEST_LLM_COMPLETION":
            # The "Think" step
            prompt = request.payload.get("prompt")
            # In a real system, you'd build a complex prompt from the state
            # llm_response = your_llm_library.generate(prompt=prompt, context=state.dict())
            llm_response = f"LLM thought based on: {prompt}" # Dummy response
            state.scratchpad.append(llm_response)
            STATE_STORE[session_id] = state
            return MCPResponse(success=True, data={"llm_response": llm_response})

        case "REQUEST_TOOL_EXECUTION":
            # The "Act" step
            tool_name = request.payload.get("tool_name")
            arguments = request.payload.get("arguments", {})
            
            if not hasattr(available_tools, tool_name):
                return MCPResponse(success=False, error=f"Tool '{tool_name}' not found.")
            
            tool_func = getattr(available_tools, tool_name)
            
            try:
                # Securely execute the tool
                result = await tool_func(**arguments)
                # Update state
                history_item = ActionHistoryItem(
                    action_type="TOOL_CALL", 
                    content=ToolCall(tool_name=tool_name, arguments=arguments),
                    result=str(result)
                )
                state.action_history.append(history_item)
                # A more advanced agent would update its world_model here
                state.world_model[f"{tool_name}_result"] = result
                STATE_STORE[session_id] = state
                return MCPResponse(success=True, data={"result": result})

            except Exception as e:
                return MCPResponse(success=False, error=f"Tool execution failed: {str(e)}")
        
        case "FINALIZE_OBJECTIVE":
            state.status = "SUCCESS"
            final_answer = request.payload.get("answer")
            state.scratchpad.append(f"Final Answer: {final_answer}")
            STATE_STORE[session_id] = state
            return MCPResponse(success=True, data={"message": "Objective finalized."})

    return MCPResponse(success=False, error="Invalid MCP action")