Agentic Coding
Fitness

Comparing Prompt Patterns

Open Claude.ai, ChatGPT, and Gemini side by side. We'll test the same task with different prompting strategies.

Task: Analyze an energy bill for savings opportunities

ZERO-SHOT
This building uses 45,000 kWh/month. HVAC is 40%, lighting 25%,
equipment 35%. Electricity costs 4.5 THB/kWh.
What are the top 3 energy savings opportunities?

FEW-SHOT
Example 1:
Building: Office 500sqm, 30,000 kWh/month, HVAC 45%
Analysis: Replace split AC with VRF system → save 25% HVAC = 3,375 kWh
Savings: 3,375 × 4.5 = 15,187 THB/month

Example 2:
Building: Retail 200sqm, 15,000 kWh/month, Lighting 35%
Analysis: Switch to LED → save 60% lighting = 3,150 kWh
Savings: 3,150 × 4.5 = 14,175 THB/month

Now analyze:
Building: Hotel 2,000sqm, 45,000 kWh/month
HVAC 40%, Lighting 25%, Equipment 35%
Electricity: 4.5 THB/kWh

CHAIN-OF-THOUGHT
A hotel uses 45,000 kWh/month. HVAC is 40%, lighting 25%,
equipment 35%. Electricity costs 4.5 THB/kWh.

Think step by step:
1. Calculate kWh for each category
2. Identify realistic % reduction for each
3. Calculate kWh saved and THB saved per month
4. Rank by ROI (payback period)
5. Give specific technology recommendations

ROLE-BASED
You are a certified energy auditor (CEA) with 15 years of experience
auditing commercial buildings in Southeast Asia. You specialize in
tropical climate HVAC optimization.

Given this building profile:
- Type: Hotel, 2,000 sqm, Bangkok
- Monthly consumption: 45,000 kWh
- Breakdown: HVAC 40%, Lighting 25%, Equipment 35%
- Rate: 4.5 THB/kWh (TOU peak/off-peak)
- Operating hours: 24/7

Provide your professional audit findings with:
- Specific equipment recommendations (brands available in Thailand)
- Expected payback periods
- Implementation priority order

Build a Prompt Library for 5 Real Tasks

"You are a senior building engineer with 20 years of experience
in tropical climate HVAC systems. Analyze this energy data..."

"A building uses 500kWh/day. If HVAC is 40% and we reduce it by 20%,
how much do we save? Think step by step."

Your First Claude API Call

PYTHON — basic_call.py
import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env

# === Basic single message ===
response = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Explain agentic AI in 3 sentences."}
    ]
)
print(response.content[0].text)
print(f"\nTokens: {response.usage.input_tokens} in, {response.usage.output_tokens} out")

PYTHON — streaming.py
import anthropic

client = anthropic.Anthropic()

# === Streaming — tokens appear in real-time ===
print("AI: ", end="")
with client.messages.stream(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Write a haiku about coding agents."}
    ]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
print()  # newline at end

PYTHON — multi_turn.py
import anthropic

client = anthropic.Anthropic()
messages = []

def chat(user_msg):
    messages.append({"role": "user", "content": user_msg})
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=1024,
        system="You are a helpful coding mentor. Be concise.",
        messages=messages
    )
    assistant_msg = response.content[0].text
    messages.append({"role": "assistant", "content": assistant_msg})
    return assistant_msg

# Multi-turn conversation
print(chat("What is a Python decorator?"))
print(chat("Show me a simple example."))
print(chat("Now show me a decorator with arguments."))

Build a Complete AI Chat CLI

PYTHON — starter: chat_cli.py
import anthropic
import time

client = anthropic.Anthropic()
messages = []
total_input_tokens = 0
total_output_tokens = 0
current_model = "claude-sonnet-4-5-20250514"

MODEL_MAP = {
    "haiku": "claude-haiku-4-5-20251001",
    "sonnet": "claude-sonnet-4-5-20250514",
    "opus": "claude-opus-4-5-20250514",
}

def call_with_retry(messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model=current_model,
                max_tokens=2048,
                system="You are a helpful AI coding assistant.",
                messages=messages
            )
        except anthropic.RateLimitError:
            wait = 2 ** attempt
            print(f"  [Rate limited, retrying in {wait}s...]")
            time.sleep(wait)
        except anthropic.APIError as e:
            print(f"  [API error: {e}]")
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

print("=== Agentic Coding Fitness — AI Chat CLI ===")
print("Commands: /model [haiku|sonnet|opus], /cost, /quit\n")

while True:
    user_input = input("You: ").strip()
    if not user_input:
        continue
    if user_input == "/quit":
        break
    if user_input == "/cost":
        print(f"  Tokens: {total_input_tokens} in, {total_output_tokens} out")
        continue
    if user_input.startswith("/model "):
        model_name = user_input.split()[1]
        if model_name in MODEL_MAP:
            current_model = MODEL_MAP[model_name]
            print(f"  Switched to {model_name}")
        continue

    messages.append({"role": "user", "content": user_input})
    response = call_with_retry(messages)
    reply = response.content[0].text
    messages.append({"role": "assistant", "content": reply})
    
    total_input_tokens += response.usage.input_tokens
    total_output_tokens += response.usage.output_tokens
    
    print(f"\nClaude: {reply}")
    print(f"  [{response.usage.input_tokens}+{response.usage.output_tokens} tokens]\n")

messages = []
def chat(user_msg):
    messages.append({"role": "user", "content": user_msg})
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=1024, messages=messages
    )
    assistant_msg = response.content[0].text
    # What line goes here?
    return assistant_msg

response = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    messages=[{"role": "user", "content": "Hello"}]
)

PYTHON — tool_use_demo.py
import anthropic
import json

client = anthropic.Anthropic()

# === Define tools ===
tools = [
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression. Use for any math.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "The math expression to evaluate, e.g. '2 + 2' or '500000 * 0.2'"
                }
            },
            "required": ["expression"]
        }
    },
    {
        "name": "get_weather",
        "description": "Get current weather for a city.",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name, e.g. 'Bangkok'"}
            },
            "required": ["city"]
        }
    }
]

# === Tool implementations ===
def execute_tool(name, inputs):
    if name == "calculate":
        try:
            result = eval(inputs["expression"])  # ⚠️ Use ast.literal_eval in production!
            return str(result)
        except Exception as e:
            return f"Error: {e}"
    elif name == "get_weather":
        # Simulated — replace with real API
        weather_data = {"Bangkok": "32°C, Humid, Partly Cloudy",
                        "Singapore": "30°C, Thunderstorms"}
        return weather_data.get(inputs["city"], "Weather data not available")

# === Conversation loop with tool use ===
def ask(question):
    messages = [{"role": "user", "content": question}]
    
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        
        # Check if Claude wants to use tools
        tool_calls = [b for b in response.content if b.type == "tool_use"]
        
        if not tool_calls:
            # No tool calls — return text response
            return response.content[0].text
        
        # Execute each tool call
        messages.append({"role": "assistant", "content": response.content})
        
        for tool_call in tool_calls:
            print(f"  🔧 Using tool: {tool_call.name}({tool_call.input})")
            result = execute_tool(tool_call.name, tool_call.input)
            messages.append({
                "role": "user",
                "content": [{"type": "tool_result",
                             "tool_use_id": tool_call.id,
                             "content": result}]
            })

# Test it!
print(ask("What's the weather in Bangkok and what's 45000 * 4.5?"))
print(ask("If Bangkok is 32°C, what is that in Fahrenheit?"))

Build a Smart Assistant with 3 Tools

A. Claude returns tool_use block
B. Your code sends tool_result back to Claude
C. You define tools with JSON schemas
D. Claude incorporates result into final answer
E. Your code executes the tool function
F. User asks a question requiring external data

messages.append({
    "role": "user",
    "content": [{"type": "tool_result", "content": "32°C, Sunny"}]
})

PYTHON — research_pipeline.py
import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic()

class ResearchPipeline:
    def __init__(self):
        self.state = {
            "topic": "",
            "queries": [],
            "sources": [],
            "summaries": [],
            "report": "",
            "quality_score": 0,
            "log": []
        }
    
    def _log(self, step, msg):
        entry = {"step": step, "time": datetime.now().isoformat(), "msg": msg}
        self.state["log"].append(entry)
        print(f"  [{step}] {msg}")
    
    def _ask(self, prompt, max_tokens=1024):
        """Helper: single Claude call"""
        resp = client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )
        return resp.content[0].text
    
    def step1_generate_queries(self, topic):
        """Generate 3 search queries for the topic"""
        self.state["topic"] = topic
        self._log("QUERIES", f"Generating queries for: {topic}")
        
        result = self._ask(f"""Generate exactly 3 search queries to research: "{topic}"
Return as JSON array: ["query1", "query2", "query3"]
Only return the JSON, nothing else.""")
        
        self.state["queries"] = json.loads(result)
        self._log("QUERIES", f"Generated: {self.state['queries']}")
    
    def step2_search(self):
        """Simulate searching (replace with real search API)"""
        for q in self.state["queries"]:
            self._log("SEARCH", f"Searching: {q}")
            # Simulate search results — replace with real API
            source = self._ask(f"""Pretend you are a search engine.
For the query "{q}", provide a 200-word factual article excerpt.
Include specific statistics and named sources where possible.""")
            self.state["sources"].append({"query": q, "content": source})
    
    def step3_summarize(self):
        """Summarize each source"""
        for src in self.state["sources"]:
            self._log("SUMMARIZE", f"Summarizing source for: {src['query']}")
            summary = self._ask(f"""Summarize this in 3 key bullet points:
{src['content']}
Format: - Key point 1\n- Key point 2\n- Key point 3""")
            self.state["summaries"].append(summary)
    
    def step4_synthesize(self):
        """Combine all summaries into final report"""
        self._log("SYNTHESIZE", "Creating final report...")
        all_summaries = "\n\n".join(
            [f"Source {i+1}:\n{s}" for i, s in enumerate(self.state["summaries"])]
        )
        self.state["report"] = self._ask(f"""You are a research analyst.
Synthesize these findings into a coherent 300-word report on "{self.state['topic']}":

{all_summaries}

Structure: Introduction → Key Findings → Implications → Conclusion""")
    
    def step5_quality_score(self):
        """Rate the report quality"""
        self._log("QA", "Scoring report quality...")
        score_result = self._ask(f"""Rate this research report 1-10 for:
- Accuracy (are claims supported?)
- Completeness (are key aspects covered?)
- Clarity (is it well-written?)
- Usefulness (would someone act on this?)

Report:
{self.state['report']}

Return as JSON: {{"accuracy": N, "completeness": N, "clarity": N, "usefulness": N, "overall": N, "feedback": "..."}}""")
        self.state["quality_score"] = json.loads(score_result)
    
    def run(self, topic):
        """Execute the full pipeline"""
        print(f"\n{'='*60}")
        print(f"RESEARCH PIPELINE: {topic}")
        print(f"{'='*60}\n")
        
        self.step1_generate_queries(topic)
        self.step2_search()
        self.step3_summarize()
        self.step4_synthesize()
        self.step5_quality_score()
        
        print(f"\n{'='*60}")
        print("FINAL REPORT:")
        print(f"{'='*60}")
        print(self.state["report"])
        print(f"\nQuality Score: {self.state['quality_score']}")
        return self.state

# RUN IT
pipeline = ResearchPipeline()
result = pipeline.run("AI-powered building energy optimization in Southeast Asia")

self.step4_synthesize()
self.step5_quality_score()
# Add branching logic here

PYTHON — agent.py — The Core Agent Framework
import anthropic
import json
import subprocess
import os

client = anthropic.Anthropic()

class Agent:
    def __init__(self, system_prompt, tools, tool_executor, max_iterations=10):
        self.system_prompt = system_prompt
        self.tools = tools
        self.tool_executor = tool_executor
        self.max_iterations = max_iterations
        self.messages = []
        self.iteration = 0
    
    def run(self, goal):
        """The core agent loop"""
        print(f"\n🎯 Agent Goal: {goal}\n")
        self.messages = [{"role": "user", "content": goal}]
        
        for i in range(self.max_iterations):
            self.iteration = i + 1
            print(f"--- Iteration {self.iteration} ---")
            
            # REASON + ACT: Ask Claude what to do
            response = client.messages.create(
                model="claude-sonnet-4-5-20250514",
                max_tokens=4096,
                system=self.system_prompt,
                tools=self.tools,
                messages=self.messages
            )
            
            # Check response
            has_tool_use = any(b.type == "tool_use" for b in response.content)
            text_blocks = [b.text for b in response.content if b.type == "text"]
            
            # Print any reasoning
            for text in text_blocks:
                print(f"  💭 {text[:200]}")
            
            # DONE check: if stop_reason is "end_turn" and no tool calls
            if response.stop_reason == "end_turn" and not has_tool_use:
                print(f"\n✅ Agent finished in {self.iteration} iterations")
                return text_blocks[-1] if text_blocks else "Done"
            
            # EXECUTE tool calls
            self.messages.append({"role": "assistant", "content": response.content})
            tool_results = []
            
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  🔧 {block.name}({json.dumps(block.input)[:100]})")
                    result = self.tool_executor(block.name, block.input)
                    print(f"  📋 Result: {str(result)[:150]}")
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result)
                    })
            
            self.messages.append({"role": "user", "content": tool_results})
        
        print(f"\n⚠️ Max iterations ({self.max_iterations}) reached")
        return "Max iterations reached"

# === CODE REVIEW AGENT ===
code_review_tools = [
    {
        "name": "read_file",
        "description": "Read contents of a Python file",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "Write content to a file",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    },
    {
        "name": "run_python",
        "description": "Run a Python file and return stdout/stderr",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        }
    },
    {
        "name": "run_lint",
        "description": "Run flake8 linter on a Python file",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        }
    }
]

def execute_code_tool(name, inputs):
    if name == "read_file":
        try:
            with open(inputs["path"]) as f:
                return f.read()
        except FileNotFoundError:
            return f"Error: File not found: {inputs['path']}"
    elif name == "write_file":
        with open(inputs["path"], "w") as f:
            f.write(inputs["content"])
        return f"Written to {inputs['path']}"
    elif name == "run_python":
        result = subprocess.run(
            ["python", inputs["path"]],
            capture_output=True, text=True, timeout=10
        )
        return f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}\nReturn code: {result.returncode}"
    elif name == "run_lint":
        result = subprocess.run(
            ["python", "-m", "flake8", inputs["path"]],
            capture_output=True, text=True
        )
        return result.stdout or "No lint issues found!"

# Create the agent
agent = Agent(
    system_prompt="""You are a code review agent. Your process:
1. Read the target file
2. Run the linter to find issues  
3. Analyze the code for bugs, style issues, and improvements
4. Write a fixed version of the file
5. Run the file to verify it works
6. If there are errors, fix them and try again
When everything is clean and working, explain what you fixed.""",
    tools=code_review_tools,
    tool_executor=execute_code_tool,
    max_iterations=10
)

# Run on a test file
result = agent.run("Review and fix the file 'sample.py'. Fix all bugs and style issues.")

Build a Complete REST API with Claude Code

# Project: Task Manager API
## Stack: FastAPI, Python 3.11, SQLite, pytest
## Conventions: type hints everywhere, docstrings on all public functions
## Testing: pytest with 80%+ coverage target
## Auth: JWT tokens

Build a Project Dashboard Agent with MCP

PYTHON — custom_mcp_server.py (simplified)
from mcp.server import Server
from mcp.types import Tool, TextContent
import json

server = Server("project-dashboard")

@server.tool()
async def get_team_status():
    """Get current team member status and availability"""
    return TextContent(
        type="text",
        text=json.dumps({
            "team_size": 8,
            "available": 6,
            "on_leave": ["Alice", "Bob"],
            "sprint": "Sprint 14",
            "days_remaining": 5
        })
    )

@server.tool()
async def get_deployment_status():
    """Check latest deployment status"""
    return TextContent(
        type="text",
        text=json.dumps({
            "environment": "production",
            "version": "2.4.1",
            "deployed_at": "2026-03-24T09:30:00Z",
            "status": "healthy",
            "uptime": "99.97%"
        })
    )

PYTHON — rag_agent.py — Starter Code
# pip install llama-index llama-index-llms-anthropic chromadb

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core import Settings
from llama_index.llms.anthropic import Anthropic

# Configure LLM
Settings.llm = Anthropic(model="claude-sonnet-4-5-20250514")

# Step 1: Load documents from a folder
documents = SimpleDirectoryReader("./company_docs").load_data()
print(f"Loaded {len(documents)} documents")

# Step 2: Build the vector index (auto-chunks, embeds, stores)
index = VectorStoreIndex.from_documents(documents)

# Step 3: Create a query engine
query_engine = index.as_query_engine(similarity_top_k=5)

# Step 4: Ask questions!
response = query_engine.query(
    "What is our company's policy on remote work?"
)
print(response)
print(f"\nSources: {[n.metadata['file_name'] for n in response.source_nodes]}")

PYTHON — content_crew.py
# pip install crewai crewai-tools

from crewai import Agent, Task, Crew, Process

# === Define Agents ===
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive, accurate, current information on the topic",
    backstory="""You are a meticulous research analyst who cross-references
    multiple sources. You focus on recent developments and data-backed insights.
    For SE Asia topics, you prioritize regional sources and local context.""",
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role="Technical Content Writer",
    goal="Write engaging, well-structured blog posts that educate and inspire",
    backstory="""You are an award-winning tech blogger who makes complex topics
    accessible. Your writing is clear, uses real examples, and includes
    actionable takeaways. You write for a developer audience in Southeast Asia.""",
    verbose=True,
    allow_delegation=False
)

editor = Agent(
    role="Senior Editor",
    goal="Ensure content is polished, accurate, and impactful",
    backstory="""You are a senior editor at a major tech publication.
    You check facts, improve clarity, fix structure issues, and ensure
    the piece delivers on its promise. You are constructively critical.""",
    verbose=True,
    allow_delegation=False
)

# === Define Tasks ===
research_task = Task(
    description="""Research the topic: {topic}
    
    Find: key trends, statistics, real-world examples, expert opinions.
    Focus on developments from the last 6 months.
    Include at least 3 specific data points or statistics.
    Output a structured research brief with sections.""",
    expected_output="A 500-word research brief with sourced data points",
    agent=researcher
)

writing_task = Task(
    description="""Using the research brief, write a blog post on: {topic}
    
    Requirements:
    - 800-1000 words
    - Engaging title and subtitle
    - Introduction with a hook
    - 3-4 main sections with headers
    - Real examples or case studies
    - Actionable conclusion
    - Write for developers in Southeast Asia""",
    expected_output="A complete, well-structured blog post",
    agent=writer,
    context=[research_task]  # Gets output from research
)

editing_task = Task(
    description="""Review and improve the blog post.
    
    Check for:
    - Factual accuracy (cross-reference with research brief)
    - Clarity and readability
    - Structure and flow
    - Grammar and style
    - Actionability of conclusions
    
    Return the final polished version with your editorial notes.""",
    expected_output="Final polished blog post ready for publication",
    agent=editor,
    context=[research_task, writing_task],
    human_input=True  # Ask human for approval before finalizing
)

# === Create and Run Crew ===
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(inputs={
    "topic": "How Agentic AI is Transforming Building Management in Southeast Asia"
})

print("\n" + "="*60)
print("FINAL OUTPUT:")
print("="*60)
print(result)

PYTHON — support_graph.py
# pip install langgraph langchain-anthropic

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-5-20250514")

# === Define State ===
class SupportState(TypedDict):
    ticket: str
    category: str
    response: str
    confidence: float
    needs_escalation: bool
    qa_approved: bool

# === Node Functions ===
def classify_ticket(state: SupportState) -> SupportState:
    """Router: classify the ticket"""
    result = llm.invoke(f"""Classify this support ticket into one category:
    TECHNICAL, BILLING, GENERAL
    
    Also rate your confidence 0-1.
    
    Ticket: {state['ticket']}
    
    Return JSON: {{"category": "...", "confidence": 0.X}}""")
    import json
    data = json.loads(result.content)
    return {**state, "category": data["category"], "confidence": data["confidence"]}

def handle_technical(state: SupportState) -> SupportState:
    result = llm.invoke(f"""You are a technical support specialist.
    Resolve this issue: {state['ticket']}
    Provide step-by-step troubleshooting.""")
    return {**state, "response": result.content}

def handle_billing(state: SupportState) -> SupportState:
    result = llm.invoke(f"""You are a billing specialist.
    Resolve this billing issue: {state['ticket']}
    Be empathetic and offer concrete solutions.""")
    return {**state, "response": result.content}

def handle_general(state: SupportState) -> SupportState:
    result = llm.invoke(f"""You are a customer support agent.
    Help with this request: {state['ticket']}""")
    return {**state, "response": result.content}

def qa_review(state: SupportState) -> SupportState:
    result = llm.invoke(f"""Review this support response for quality:
    
    Original ticket: {state['ticket']}
    Response: {state['response']}
    
    Is this response helpful, accurate, and professional? (yes/no)
    If no, what needs improvement?""")
    approved = "yes" in result.content.lower()[:50]
    return {**state, "qa_approved": approved}

# === Routing Logic ===
def route_by_category(state: SupportState) -> str:
    if state["confidence"] < 0.7:
        return "escalate"
    return state["category"].lower()

def route_after_qa(state: SupportState) -> str:
    return "end" if state["qa_approved"] else "escalate"

# === Build the Graph ===
graph = StateGraph(SupportState)

graph.add_node("classify", classify_ticket)
graph.add_node("technical", handle_technical)
graph.add_node("billing", handle_billing)
graph.add_node("general", handle_general)
graph.add_node("qa", qa_review)

graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_by_category, {
    "technical": "technical",
    "billing": "billing",
    "general": "general",
    "escalate": END
})
graph.add_edge("technical", "qa")
graph.add_edge("billing", "qa")
graph.add_edge("general", "qa")
graph.add_conditional_edges("qa", route_after_qa, {
    "end": END,
    "escalate": END
})

app = graph.compile()

# === Test it ===
result = app.invoke({
    "ticket": "My HVAC controller is showing error code E47 and the system won't start",
    "category": "", "response": "", "confidence": 0.0,
    "needs_escalation": False, "qa_approved": False
})
print(f"Category: {result['category']} (confidence: {result['confidence']})")
print(f"QA Approved: {result['qa_approved']}")
print(f"Response: {result['response'][:500]}")

PYTHON — audit_swarm.py
import anthropic
import asyncio
import json
import time

client = anthropic.Anthropic()

# === Specialist Agent Prompts ===
SPECIALISTS = {
    "security": {
        "name": "Security Auditor",
        "prompt": """Analyze this code for security vulnerabilities:
- SQL injection, XSS, CSRF
- Hardcoded secrets or credentials
- Insecure dependencies
- Missing input validation
- Authentication/authorization flaws
Return JSON: {"findings": [{"severity": "critical|high|medium|low", "issue": "...", "line": N, "fix": "..."}]}"""
    },
    "performance": {
        "name": "Performance Analyst",
        "prompt": """Analyze this code for performance issues:
- N+1 queries, missing indexes
- Memory leaks or excessive allocation  
- Blocking operations in async code
- Missing caching opportunities
- Inefficient algorithms (O(n²) when O(n) possible)
Return JSON: {"findings": [...]}"""
    },
    "style": {
        "name": "Code Style Reviewer",
        "prompt": """Review code style and best practices:
- PEP 8 compliance
- Type hints usage
- Docstring completeness
- Naming conventions
- Code complexity (functions too long?)
Return JSON: {"findings": [...]}"""
    },
    "testing": {
        "name": "Test Coverage Analyst",
        "prompt": """Analyze test coverage and quality:
- Which functions lack tests?
- Are edge cases covered?
- Are error paths tested?
- Test naming and organization
- Missing integration tests
Return JSON: {"findings": [...]}"""
    },
    "docs": {
        "name": "Documentation Reviewer",
        "prompt": """Review documentation completeness:
- README accuracy and completeness
- API documentation
- Inline comments quality
- Architecture documentation
- Setup/deployment guides
Return JSON: {"findings": [...]}"""
    }
}

async def run_specialist(name, spec, code):
    """Run one specialist agent"""
    start = time.time()
    response = client.messages.create(
        model="claude-sonnet-4-5-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": f"{spec['prompt']}\n\nCode:\n```\n{code}\n```"}]
    )
    elapsed = time.time() - start
    print(f"  ✅ {spec['name']} finished in {elapsed:.1f}s")
    try:
        return {"specialist": name, "results": json.loads(response.content[0].text)}
    except json.JSONDecodeError:
        return {"specialist": name, "results": {"raw": response.content[0].text}}

async def run_swarm(code):
    """Run all specialists in parallel"""
    print("🐝 Launching audit swarm...\n")
    start = time.time()
    
    tasks = [
        run_specialist(name, spec, code) 
        for name, spec in SPECIALISTS.items()
    ]
    results = await asyncio.gather(*tasks)
    
    total = time.time() - start
    print(f"\n⏱ All {len(results)} agents finished in {total:.1f}s")
    return results

def synthesize_report(results):
    """Orchestrator: combine all findings into unified report"""
    all_findings = []
    for r in results:
        if "findings" in r.get("results", {}):
            for f in r["results"]["findings"]:
                f["source"] = r["specialist"]
                all_findings.append(f)
    
    # Sort by severity
    severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
    all_findings.sort(key=lambda f: severity_order.get(f.get("severity", "low"), 4))
    
    print(f"\n{'='*60}")
    print(f"UNIFIED AUDIT REPORT — {len(all_findings)} findings")
    print(f"{'='*60}")
    for f in all_findings:
        icon = {"critical":"🔴","high":"🟠","medium":"🟡","low":"🔵"}.get(f.get("severity"),"⚪")
        print(f"{icon} [{f.get('severity','?').upper()}] [{f['source']}] {f.get('issue','')}")
    
    return all_findings

# === RUN ===
sample_code = open("your_project.py").read()  # or paste code inline
results = asyncio.run(run_swarm(sample_code))
report = synthesize_report(results)

Smart Building Energy Optimizer

Teams of 3–4 build a multi-agent building management system. Each team member owns one agent.

Production-Harden Your Building Optimizer

PYTHON — governance.py — Audit Logger
import json
import time
from datetime import datetime
from functools import wraps

class AuditLogger:
    def __init__(self, log_file="audit.jsonl"):
        self.log_file = log_file
        self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.total_cost = 0
        self.cost_limit = 5.00  # USD per session
    
    def log(self, agent_id, action, details, tokens=0, cost=0):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "session": self.session_id,
            "agent": agent_id,
            "action": action,
            "details": details,
            "tokens": tokens,
            "cost_usd": round(cost, 4),
            "cumulative_cost": round(self.total_cost + cost, 4)
        }
        self.total_cost += cost
        
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")
        
        # ALERT if cost limit approaching
        if self.total_cost > self.cost_limit * 0.8:
            print(f"⚠️ COST ALERT: ${self.total_cost:.2f} / ${self.cost_limit:.2f}")
        if self.total_cost > self.cost_limit:
            raise Exception(f"🚫 COST LIMIT EXCEEDED: ${self.total_cost:.2f}")
        
        return entry

# Usage: wrap your agent tool calls
audit = AuditLogger()
audit.log("hvac_optimizer", "tool_call", {"tool": "set_temperature", "value": 24}, tokens=150, cost=0.002)

Capstone Project Ideas

Design Document Template

Debugging Tips for Multi-Agent Systems

Awards