undefined

undefined

undefined

for step in range(max_steps):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )
    
    message = response.choices[0].message
    messages.append(message)
    
    # Check if done
    if not message.tool_calls:
        return message.content
    
    # Execute tool calls
    for tool_call in message.tool_calls:
        if tool_call.function.name == "calculate":
            import json
            args = json.loads(tool_call.function.arguments)
            result = calculate(args["expression"])
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

return "Max steps reached"

**Deliverable**: 50-150 line agent that can choose tools, execute them, and return final answer.

undefined

**Deliverable**: Research assistant that searches, filters, summarizes, and outputs citations.

undefined

def register(self, name: str, func: callable, requires_approval: bool = False):
    """Register tool with optional approval gate."""
    self.tools[name] = func
    self.permissions[name] = {
        "requires_approval": requires_approval,
        "allowed_domains": []  # Could expand to domain restrictions
    }

def execute(self, name: str, args: dict, auto_approve: bool = False):
    """Execute tool with permission check."""
    if name not in self.tools:
        raise ValueError(f"Tool {name} not found")
    
    if self.permissions[name]["requires_approval"] and not auto_approve:
        # In real system, this would trigger user confirmation
        print(f"⚠️  Tool {name} requires approval. Args: {args}")
        confirm = input("Approve? (y/n): ")
        if confirm.lower() != 'y':
            return "Tool execution denied by user"
    
    return self.tools[name](**args)

**Deliverable**: Working agent harness demo with README, example runs, and failure logs.

undefined

**Deliverable**: Multi-agent system with clear roles (e.g., research → write → review → revise).

undefined

**MCP Server Example**:

```typescript
// MCP server for custom tools
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  {
    name: "my-tools-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// Register tool
server.setRequestHandler("tools/list", async () => {
  return {
    tools: [
      {
        name: "search_codebase",
        description: "Search company codebase using semantic search",
        inputSchema: {
          type: "object",
          properties: {
            query: { type: "string" },
            language: { type: "string", enum: ["python", "typescript", "all"] }
          },
          required: ["query"]
        }
      }
    ]
  };
});

server.setRequestHandler("tools/call", async (request) => {
  if (request.params.name === "search_codebase") {
    const { query, language } = request.params.arguments;
    // Implement search logic
    return {
      content: [{ type: "text", text: `Results for ${query}...` }]
    };
  }
});

// Start server
const transport = new StdioServerTransport();
await server.connect(transport);

undefined

result = await agent.run()
print(result)

**Anthropic Computer Use Example**:

```python
import anthropic
import os

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    tools=[
        {
            "type": "computer_20241022",
            "name": "computer",
            "display_width_px": 1024,
            "display_height_px": 768,
        }
    ],
    messages=[
        {
            "role": "user",
            "content": "Open a browser and search for 'AI agent frameworks'"
        }
    ]
)

print(response)

undefined

def run_evaluation(self) -> Dict:
    """Run all test cases and collect metrics."""
    for test in self.test_cases:
        start_time = datetime.now()
        
        try:
            result = self.agent_fn(test.input, max_steps=test.max_steps)
            success = self._check_success(result, test.expected_output)
            
            latency = (datetime.now() - start_time).total_seconds() * 1000
            
            eval_result = EvalResult(
                test_id=test.id,
                success=success,
                actual_output=result["output"],
                steps_taken=result["steps"],
                cost_usd=result["cost"],
                latency_ms=latency,
                trace=result.get("trace")
            )
        except Exception as e:
            eval_result = EvalResult(
                test_id=test.id,
                success=False,
                actual_output="",
                steps_taken=0,
                cost_usd=0,
                latency_ms=0,
                failure_reason=str(e)
            )
        
        self.results.append(eval_result)
    
    return self._compute_metrics()

def _check_success(self, result: Dict, expected: str) -> bool:
    """Check if output matches expected (implement your logic)."""
    return expected.lower() in result["output"].lower()

def _compute_metrics(self) -> Dict:
    """Aggregate metrics across all tests."""
    total = len(self.results)
    successful = sum(1 for r in self.results if r.success)
    
    return {
        "success_rate": successful / total if total > 0 else 0,
        "total_tests": total,
        "total_cost_usd": sum(r.cost_usd for r in self.results),
        "avg_latency_ms": sum(r.latency_ms for r in self.results) / total if total > 0 else 0,
        "failure_reasons": [r.failure_reason for r in self.results if not r.success]
    }

def export_report(self, filename: str):
    """Export detailed report as JSON."""
    report = {
        "metrics": self._compute_metrics(),
        "results": [
            {
                "test_id": r.test_id,
                "success": r.success,
                "steps": r.steps_taken,
                "cost": r.cost_usd,
                "latency_ms": r.latency_ms,
                "failure_reason": r.failure_reason
            }
            for r in self.results
        ]
    }
    
    with open(filename, "w") as f:
        json.dump(report, f, indent=2)

**Safety Patterns**:

```python

    approval = input("\nApprove this action? (yes/no): ")
    if approval.lower() != "yes":
        return {"status": "rejected", "reason": "User denied approval"}

# Execute tool
return tools_registry[tool_name](**args)

**Deliverable**: Eval suite with fixed test set, success rate tracking, and cost/latency metrics.

undefined

def add_message(self, message: dict):
    """Add message and compact if needed."""
    self.messages.append(message)
    
    # Estimate tokens (rough: 1 token ≈ 4 chars)
    total_chars = sum(len(str(m)) for m in self.messages)
    estimated_tokens = total_chars // 4
    
    if estimated_tokens > self.max_tokens:
        self._compact()

def _compact(self):
    """Keep system message, last user message, and recent history."""
    system = [m for m in self.messages if m["role"] == "system"]
    recent = self.messages[-10:]  # Keep last 10 messages
    
    # Summarize middle messages (in production, use LLM)
    if len(self.messages) > 12:
        summary = {
            "role": "system",
            "content": f"[Previous conversation summarized: {len(self.messages) - 12} messages]"
        }
        self.messages = system + [summary] + recent

undefined

def run_agent_with_limits(task: str, max_steps: int = 10, max_cost: float = 1.0):
    total_cost = 0.0
    
    for step in range(max_steps):
        if total_cost > max_cost:
            return {"error": "Cost limit exceeded", "partial_result": current_state}
        
        # Run agent step
        result = agent.step()
        total_cost += result.cost
        
        if result.is_final:
            return result
    
    return {"error": "Max steps exceeded", "partial_result": current_state}

from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(..., min_length=3, description="Search query, at least 3 characters")
    max_results: int = Field(5, ge=1, le=20, description="Number of results, 1-20")

def search_tool(input: SearchInput) -> dict:
    """Type-safe search tool."""
    # Validation happens automatically via Pydantic
    return {"results": [...]}

def verify_citations(text: str, sources: List[str]) -> bool:
    """Check if all citations in text exist in sources."""
    import re
    
    cited = re.findall(r'\[(\d+)\]', text)
    max_source_idx = len(sources) - 1
    
    for citation in cited:
        if int(citation) > max_source_idx:
            return False
    
    return True