AI Agents in Software Engineering: Architecture Patterns That Actually Work

Observe → Think → Act → Observe (loop until goal met or budget exceeded)

import anthropic
import json
from typing import Any

client = anthropic.Anthropic()

# Define the tools the agent can use
TOOLS = [
    {
        "name": "read_file",
        "description": "Read the contents of a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path to read"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "run_command",
        "description": "Run a shell command and return stdout/stderr",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {"type": "string", "description": "Shell command to run"}
            },
            "required": ["command"]
        }
    },
    {
        "name": "write_file",
        "description": "Write content to a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "File path"},
                "content": {"type": "string", "description": "Content to write"}
            },
            "required": ["path", "content"]
        }
    }
]

def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return its output as a string."""
    import subprocess
    from pathlib import Path
    
    if tool_name == "read_file":
        path = Path(tool_input["path"])
        if not path.exists():
            return f"Error: file {path} does not exist"
        return path.read_text()
    
    elif tool_name == "run_command":
        result = subprocess.run(
            tool_input["command"],
            shell=True,
            capture_output=True,
            text=True,
            timeout=30
        )
        output = result.stdout
        if result.stderr:
            output += f"\nSTDERR:\n{result.stderr}"
        return output or "(no output)"
    
    elif tool_name == "write_file":
        path = Path(tool_input["path"])
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(tool_input["content"])
        return f"Written {len(tool_input['content'])} bytes to {path}"
    
    return f"Unknown tool: {tool_name}"

def run_agent(task: str, max_iterations: int = 20) -> str:
    """Run an agent loop until task completion or iteration limit."""
    messages = [{"role": "user", "content": task}]
    
    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=TOOLS,
            messages=messages,
        )
        
        # Add assistant response to history
        messages.append({"role": "assistant", "content": response.content})
        
        # Check if agent is done
        if response.stop_reason == "end_turn":
            # Extract text response
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "Task completed"
        
        # Process tool calls
        if response.stop_reason == "tool_use":
            tool_results = []
            
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  → Calling {block.name}({json.dumps(block.input)[:100]})")
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            
            messages.append({"role": "user", "content": tool_results})
    
    return f"Reached iteration limit ({max_iterations})"

Agent ←→ MCP Client ←→ MCP Server (filesystem, GitHub, databases, etc.)

# Example: connecting an agent to an MCP server
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_agent_with_mcp(task: str) -> str:
    # Connect to a filesystem MCP server
    server_params = StdioServerParameters(
        command="uvx",
        args=["mcp-server-filesystem", "/workspace"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # List available tools from the MCP server
            tools_result = await session.list_tools()
            tools = [t.model_dump() for t in tools_result.tools]
            
            # Run agent with MCP tools
            # (same loop as above, but tools come from MCP server)
            print(f"Available tools: {[t['name'] for t in tools]}")
            return "task_result"

def compress_history(messages: list, keep_last_n: int = 10) -> list:
    """Compress old messages when approaching context limit."""
    if len(messages) <= keep_last_n:
        return messages
    
    # Summarize old messages
    old_messages = messages[:-keep_last_n]
    summary_prompt = f"""Summarize these agent actions and their results concisely:
    
{json.dumps(old_messages, indent=2)}

Summary (keep all file paths, error messages, and key decisions):"""
    
    summary_response = client.messages.create(
        model="claude-haiku-4-5",  # cheaper model for summarization
        max_tokens=1000,
        messages=[{"role": "user", "content": summary_prompt}]
    )
    
    summary = summary_response.content[0].text
    return [{"role": "user", "content": f"[Earlier context summary]: {summary}"}] + messages[-keep_last_n:]

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import anthropic

embed_client = anthropic.Anthropic()

def store_observation(collection: str, text: str, metadata: dict) -> None:
    """Store an agent observation in vector memory."""
    # Get embedding
    response = embed_client.embeddings.create(
        model="voyage-3",
        input=[text]
    )
    vector = response.embeddings[0]
    
    qdrant = QdrantClient("localhost", port=6333)
    qdrant.upsert(
        collection_name=collection,
        points=[PointStruct(
            id=hash(text) % (2**32),
            vector=vector,
            payload={"text": text, **metadata}
        )]
    )

def recall_relevant(collection: str, query: str, top_k: int = 5) -> list[str]:
    """Retrieve relevant past observations for a query."""
    response = embed_client.embeddings.create(
        model="voyage-3",
        input=[query]
    )
    query_vector = response.embeddings[0]
    
    qdrant = QdrantClient("localhost", port=6333)
    results = qdrant.search(
        collection_name=collection,
        query_vector=query_vector,
        limit=top_k
    )
    
    return [r.payload["text"] for r in results]

def orchestrator_loop(high_level_task: str) -> str:
    """Orchestrator that delegates to specialized sub-agents."""
    subtasks = decompose_task(high_level_task)  # LLM call
    
    results = {}
    for subtask in subtasks:
        agent_type = route_to_agent(subtask)  # LLM call
        
        if agent_type == "code_writer":
            results[subtask] = run_code_agent(subtask)
        elif agent_type == "test_runner":
            results[subtask] = run_test_agent(subtask)
        elif agent_type == "reviewer":
            results[subtask] = run_review_agent(subtask)
    
    return aggregate_results(results)  # LLM call

def generate_with_critique(task: str, max_rounds: int = 3) -> str:
    """Generate output, critique it, revise until acceptable."""
    content = run_agent(task)  # generator agent
    
    for _ in range(max_rounds):
        critique = run_agent(
            f"Review this output critically:\n\n{content}\n\n"
            f"Original task: {task}\n\n"
            "List specific issues. If acceptable, say 'APPROVED'."
        )
        
        if "APPROVED" in critique:
            return content
        
        content = run_agent(
            f"Revise based on this critique:\n\n{critique}\n\n"
            f"Current content:\n\n{content}"
        )
    
    return content

from pathlib import Path

ALLOWED_WRITE_PATHS = [Path("/workspace"), Path("/tmp/agent")]
FORBIDDEN_PATHS = [Path("/etc"), Path("/root"), Path.home() / ".ssh"]

def safe_write_file(path_str: str, content: str) -> str:
    path = Path(path_str).resolve()
    
    for forbidden in FORBIDDEN_PATHS:
        if path.is_relative_to(forbidden):
            return f"BLOCKED: cannot write to {forbidden}"
    
    if not any(path.is_relative_to(allowed) for allowed in ALLOWED_WRITE_PATHS):
        return f"BLOCKED: {path} is outside allowed write paths"
    
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    return f"Written to {path}"

import shlex

ALLOWED_COMMANDS = {"pytest", "ruff", "mypy", "npm", "git diff", "git status"}
FORBIDDEN_COMMAND_PREFIXES = ("rm -rf", "git push", "git reset --hard", "sudo", "curl", "wget")

def safe_run_command(command: str) -> str:
    # Check forbidden prefixes
    stripped = command.strip()
    for forbidden in FORBIDDEN_COMMAND_PREFIXES:
        if stripped.startswith(forbidden):
            return f"BLOCKED: command matches forbidden pattern '{forbidden}'"
    
    # Check against allowlist (optional - more restrictive)
    cmd_parts = shlex.split(stripped)
    base_cmd = cmd_parts[0] if cmd_parts else ""
    if base_cmd not in ALLOWED_COMMANDS:
        return f"BLOCKED: '{base_cmd}' not in allowed commands. Allowed: {ALLOWED_COMMANDS}"
    
    # Actually run
    import subprocess
    result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=60)
    return result.stdout + (f"\nSTDERR:\n{result.stderr}" if result.stderr else "")