Agentic AI Architecture: Memory, Tools, Control Loops & Multi-Agent Orchestration

from anthropic import Anthropic

client = Anthropic()

def run_agent(task: str, tools: list[dict], max_iterations: int = 10) -> str:
    messages = [{"role": "user", "content": task}]
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        # THINK: LLM reasons about what to do next
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )
        
        # Check stop condition:
        if response.stop_reason == "end_turn":
            # Extract text response — agent is done
            return next(b.text for b in response.content if hasattr(b, 'text'))
        
        # ACT: Extract tool calls from response
        tool_uses = [b for b in response.content if b.type == "tool_use"]
        
        if not tool_uses:
            break  # No tools called, no end — unexpected state
        
        # Append assistant's reasoning + tool calls to conversation:
        messages.append({"role": "assistant", "content": response.content})
        
        # OBSERVE: Execute each tool and collect results
        tool_results = []
        for tool_use in tool_uses:
            result = execute_tool(tool_use.name, tool_use.input)  # Your tool router
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": str(result),
            })
        
        # Feed observations back to LLM:
        messages.append({"role": "user", "content": tool_results})
    
    # Max iterations reached — return partial result with warning
    return f"[Agent reached max_iterations={max_iterations}] Last state: {messages[-1]}"

class AgentMemory:
    def __init__(self, vector_store, session_store):
        self.vector_store = vector_store   # Long-term semantic memory (Pinecone/Chroma)
        self.session_store = session_store  # Episodic memory (Redis/database)
        self.working_memory: list = []     # In-context window (messages list)
    
    def remember(self, content: str, metadata: dict):
        """Store to long-term semantic memory."""
        embedding = embed(content)
        self.vector_store.upsert(embedding, content, metadata)
    
    def recall(self, query: str, k: int = 5) -> list[str]:
        """Retrieve relevant long-term memories."""
        embedding = embed(query)
        results = self.vector_store.query(embedding, k=k)
        return [r.content for r in results]
    
    def inject_relevant_context(self, task: str) -> str:
        """Build context-enriched system prompt from long-term memory."""
        memories = self.recall(task)
        if memories:
            return f"Relevant past context:\n" + "\n".join(f"- {m}" for m in memories)
        return ""

# Tool schema (sent to the LLM):
tools = [
    {
        "name": "run_python",
        "description": "Execute Python code and return stdout/stderr. Use for calculations, data processing.",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": "Python code to execute. No imports of os, subprocess, sys."
                }
            },
            "required": ["code"]
        }
    },
    {
        "name": "web_search",
        "description": "Search the web for current information. Returns top 5 result snippets.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "max_results": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }
    }
]

# Sandboxed code execution (using E2B or Firecracker VMs):
import e2b

def execute_tool(name: str, inputs: dict) -> str:
    if name == "run_python":
        with e2b.Sandbox() as sandbox:
            # Runs in isolated microVM — cannot access host filesystem
            result = sandbox.process.start_and_wait(
                f"python3 -c '{inputs['code']}'"
            )
            return result.stdout or result.stderr
    elif name == "web_search":
        return search_api.query(inputs["query"], inputs.get("max_results", 5))
    raise ValueError(f"Unknown tool: {name}")

Without MCP:                        With MCP:
Each agent integrates tools         Agent connects to MCP servers
individually (bespoke code):        (standardised protocol):

Agent → custom Slack code           Agent → MCP Client → MCP Server (Slack)
Agent → custom DB code              Agent → MCP Client → MCP Server (PostgreSQL)
Agent → custom Calendar code        Agent → MCP Client → MCP Server (Google Calendar)

Tools are not reusable across        Any MCP-compatible agent can use
agents or frameworks                any MCP server — plug and play

# Supervisor decides which specialist to call:
AGENTS = {
    "researcher":  ResearchAgent(),   # Web search + summarisation
    "coder":       CoderAgent(),      # Code generation + execution
    "writer":      WriterAgent(),     # Long-form content generation
    "reviewer":    ReviewerAgent(),   # Quality checking + critique
}

def supervisor(task: str) -> str:
    # Supervisor LLM decides the workflow:
    plan = supervisor_llm.plan(task)  
    # e.g.: ["researcher", "coder", "reviewer"]
    
    context = task
    for agent_name in plan:
        agent = AGENTS[agent_name]
        result = agent.run(context)
        context = f"{context}\n\nAgent ({agent_name}) output:\n{result}"
    
    return context

# Each agent decides if it should transfer to another agent:
def research_agent(messages, context):
    response = llm.call(messages, tools=[web_search, transfer_to_coder])
    if response.tool == "transfer_to_coder":
        return coder_agent(messages + [response], context)
    return response.content