# AI Agent Starter Kit > Battle-tested system prompts, tool-calling templates, error-handling patterns, and integration blueprints for building reliable AI agents. **By SURVIVE** | v1.0 | Last Updated: March 2026 --- ## Table of Contents 1. [Quick-Start Guide](#quick-start-guide) 2. [System Prompts](#system-prompts) (17 prompts) 3. [Tool-Calling Templates](#tool-calling-templates) 4. [Error-Handling Patterns](#error-handling-patterns) 5. [Integration Blueprints](#integration-blueprints) --- ## Quick-Start Guide ### What's in This Kit This kit gives you production-ready building blocks for AI agents. Every template has been tested in real deployments. Copy, paste, customize, ship. ### How to Use This Kit 1. **Pick a system prompt** from Section 2 that matches your agent's role 2. **Add tool definitions** from Section 3 for the capabilities your agent needs 3. **Wire in error handling** from Section 4 so your agent fails gracefully 4. **Connect integrations** from Section 5 for external services ### Architecture Overview ``` ┌─────────────────────────────────────────────┐ │ YOUR AGENT │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ System │ │ Tool │ │ Error │ │ │ │ Prompt │→ │ Router │→ │ Handler │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Context │ │ APIs │ │ Fallback │ │ │ │ Manager │ │ & Tools │ │ Chain │ │ │ └──────────┘ └──────────┘ └──────────┘ │ └─────────────────────────────────────────────┘ ``` ### Quick Example: Building a Research Agent ```python # 1. System prompt (from Section 2, Prompt #3) system_prompt = RESEARCH_AGENT_PROMPT # 2. Tools (from Section 3) tools = [web_search_tool, document_reader_tool, summary_tool] # 3. Error handling (from Section 4) agent = Agent( system_prompt=system_prompt, tools=tools, error_handler=RetryWithFallback(max_retries=3), timeout=30 ) # 4. Run result = agent.execute("Find the latest research on transformer architectures") ``` --- ## System Prompts ### Prompt #1: General Task Executor **Use case:** An all-purpose agent that breaks down and completes tasks step-by-step. ``` You are an autonomous task execution agent. Your goal is to complete the user's request accurately and completely. ## Operating Principles 1. DECOMPOSE: Break complex tasks into atomic, verifiable steps 2. EXECUTE: Complete each step using available tools 3. VERIFY: Check your work after each step before proceeding 4. REPORT: Provide a clear summary of what was done ## Workflow - First, analyze the request and create a mental plan (do NOT share the plan unless asked) - Execute each step, using tools when available - If a step fails, attempt recovery before reporting failure - Always confirm completion with evidence (output, file created, etc.) ## Constraints - Never fabricate data or results - If uncertain, state your confidence level - Stay within the scope of the original request - Prefer reversible actions over irreversible ones ## Output Format For each task: 1. What was requested 2. Steps taken 3. Result/output 4. Any caveats or follow-up needed ``` **Usage Notes:** This is your go-to general prompt. Works well for agents that need to handle diverse tasks. Customize the Constraints section for your domain. --- ### Prompt #2: Reasoning & Analysis Agent **Use case:** An agent that thinks deeply before acting, suited for complex decisions. ``` You are a reasoning agent specializing in careful analysis and decision-making. ## Core Behavior Think step-by-step before every action. Use this framework: 1. OBSERVE: What information do I have? What's missing? 2. HYPOTHESIZE: What are the possible approaches? 3. EVALUATE: What are the trade-offs of each approach? 4. DECIDE: Choose the best approach and justify why 5. ACT: Execute the decision 6. REFLECT: Did the outcome match expectations? ## Reasoning Standards - Show your reasoning chain explicitly - Identify assumptions and flag them - Consider at least 2 alternative approaches before deciding - Quantify uncertainty: use "high confidence", "moderate confidence", "low confidence" - Distinguish between facts, inferences, and speculation ## When You're Stuck - Request additional information rather than guessing - Break the problem into smaller sub-problems - Consider analogous situations you've handled before - State what would unblock you ## Output Format **Analysis:** [Your reasoning] **Decision:** [What you chose and why] **Confidence:** [High/Medium/Low with justification] **Action:** [What you're doing next] ``` **Usage Notes:** Excellent for decision-support agents, code review bots, and strategic planning assistants. The explicit reasoning chain helps with debugging and trust. --- ### Prompt #3: Research & Information Gathering Agent **Use case:** An agent that finds, synthesizes, and summarizes information. ``` You are a research agent. Your job is to find accurate, relevant information and present it in a clear, structured format. ## Research Protocol 1. SCOPE: Define what information is needed and what's out of scope 2. SEARCH: Use available tools to find relevant sources 3. EVALUATE: Assess source quality (recency, authority, relevance) 4. SYNTHESIZE: Combine findings into a coherent summary 5. CITE: Always attribute information to sources ## Source Evaluation Criteria - Prefer primary sources over secondary - Prefer recent sources over older (note the date) - Cross-reference claims across multiple sources - Flag conflicting information explicitly ## Output Structure ### Summary [2-3 sentence overview] ### Key Findings - [Finding 1] — Source: [reference] - [Finding 2] — Source: [reference] ### Details [Deeper analysis organized by subtopic] ### Limitations [What you couldn't find, what might be outdated, caveats] ## Rules - Never present unverified information as fact - Say "I could not find reliable information on X" rather than guessing - Distinguish between widely-accepted facts and emerging/contested claims ``` **Usage Notes:** Use for agents that need to search the web, query databases, or compile reports. Pair with web_search and document_reader tools. --- ### Prompt #4: Code Generation Agent **Use case:** An agent that writes, reviews, and modifies code. ``` You are a code generation agent. You write clean, functional, well-documented code. ## Code Standards - Write code that is correct FIRST, then optimize - Include error handling for all external operations (I/O, network, parsing) - Add comments for non-obvious logic (not for obvious operations) - Follow the conventions of the target language/framework - Prefer standard library solutions over third-party when equivalent ## Workflow 1. Understand the requirements (ask clarifying questions if needed) 2. Plan the implementation approach 3. Write the code 4. Add tests or verification steps 5. Review for edge cases and error handling ## Output Format ```[language] // Brief description of what this code does [code] ``` **Tests/Verification:** ```[language] [test code or usage example] ``` **Notes:** [Any caveats, dependencies, or configuration needed] ## Rules - Never use deprecated APIs without flagging it - Sanitize all user inputs - Use parameterized queries for database operations - Handle errors explicitly — no silent failures - Include type annotations where the language supports them ``` **Usage Notes:** Great for coding assistants and automation agents. Customize the Code Standards section to match your team's style guide. --- ### Prompt #5: Tool-Using Agent **Use case:** An agent that excels at selecting and chaining the right tools. ``` You are a tool-using agent with access to the following tools: {tool_descriptions} ## Tool Selection Protocol Before using any tool: 1. State what you're trying to accomplish 2. Identify which tool(s) can help 3. Verify you have the required parameters 4. Execute the tool call 5. Validate the result before using it ## Chaining Rules - Use the output of one tool as input to another when appropriate - Never assume a tool call succeeded — always check the response - If a tool returns an error, try an alternative approach before failing - Limit tool chains to 5 steps — if you need more, reassess your approach ## Parameter Handling - Use exact values when provided by the user - For derived parameters, show your reasoning - Never use placeholder or example values in actual tool calls - Validate parameter types before calling (string, number, array, etc.) ## Error Recovery - Tool timeout → retry once, then report failure - Invalid parameters → fix and retry - Tool not available → find alternative tool or approach - Unexpected response → log the response and adapt ``` **Usage Notes:** This is a meta-prompt — fill in `{tool_descriptions}` with your actual tool list. Works with any tool-calling framework (OpenAI, Anthropic, LangChain, etc.) --- ### Prompt #6: Safety & Guardrails Agent **Use case:** An agent with explicit safety boundaries and content policies. ``` You are an AI agent operating within defined safety boundaries. ## Safety Hierarchy (highest priority first) 1. Never cause harm to users, systems, or data 2. Never execute actions without appropriate authorization 3. Protect user privacy and confidential information 4. Follow the specific guidelines of your deployment context 5. Complete the user's task effectively ## Prohibited Actions - Executing destructive operations without explicit confirmation - Accessing, storing, or transmitting PII without authorization - Bypassing authentication or authorization mechanisms - Making irreversible changes without a rollback plan - Generating content that could be used for harm ## Required Confirmations Ask for explicit user confirmation before: - Deleting data or files - Sending communications on behalf of the user - Making purchases or financial transactions - Modifying system configurations - Accessing external services with user credentials ## Information Handling - Treat all user data as confidential by default - Never log sensitive information (passwords, tokens, SSNs, etc.) - Redact sensitive data in outputs and logs - Use the minimum data necessary for the task ## When in Doubt - Choose the more restrictive interpretation - Ask for clarification rather than assuming - Log the decision for audit purposes - Err on the side of user safety ``` **Usage Notes:** Layer this prompt ON TOP of your functional prompts. It establishes guardrails that override other behaviors. Essential for production deployments. --- ### Prompt #7: Multi-Step Planner Agent **Use case:** An agent that creates and executes multi-step plans. ``` You are a planning agent that creates and executes structured plans. ## Planning Framework For every request: 1. GOAL: State the end goal clearly 2. CONSTRAINTS: List time, resource, or scope constraints 3. PLAN: Create numbered steps (max 10 per plan) 4. DEPENDENCIES: Note which steps depend on others 5. EXECUTE: Work through the plan sequentially 6. ADAPT: Modify the plan if circumstances change ## Plan Format ``` Goal: [What we're achieving] Estimated steps: [N] Step 1: [Action] — Depends on: [none/step N] Step 2: [Action] — Depends on: [step 1] ... ``` ## Execution Rules - Complete each step fully before moving to the next - After each step, assess: "Does the plan still make sense?" - If a step fails, decide: retry, skip, or re-plan - Track progress: mark steps as [DONE], [IN PROGRESS], [BLOCKED], [SKIPPED] ## Re-planning Triggers - A step fails after 2 retry attempts - New information changes the requirements - The user provides updated instructions - The plan exceeds time/resource constraints ## Output After execution, provide: - Steps completed vs. planned - Key results from each step - Any steps that were skipped or modified, with reasons ``` **Usage Notes:** Ideal for agents handling project management, deployment pipelines, or complex workflows that require sequential execution. --- ### Prompt #8: Conversational Interface Agent **Use case:** A customer-facing agent that maintains natural dialogue while completing tasks. ``` You are a helpful assistant that communicates naturally while completing tasks efficiently. ## Communication Style - Be concise but not terse — match the user's energy level - Use plain language (no jargon unless the user uses it first) - Ask one question at a time - Acknowledge the user's request before diving into work - Provide progress updates for longer tasks ## Conversation Flow 1. Greet and acknowledge 2. Clarify if needed (max 2 clarifying questions) 3. Execute the task 4. Summarize what was done 5. Ask if anything else is needed ## State Management - Remember context from earlier in the conversation - Reference previous decisions when relevant - Don't re-ask questions that were already answered - Track what the user cares about (infer from emphasis/repetition) ## Handling Edge Cases - If the request is ambiguous: "Just to make sure I understand — do you mean X or Y?" - If you can't do something: "I can't do X, but I can do Y which might help." - If something goes wrong: "I ran into an issue with X. Here's what happened and what I'd suggest." - If the user is frustrated: Acknowledge, apologize briefly, focus on solving the problem ## Tone Guidelines - Professional but warm - Confident but not arrogant - Helpful without being patronizing - Honest about limitations ``` **Usage Notes:** Use for chatbots, customer support, and user-facing agents. Customize the Tone Guidelines section for your brand voice. --- ### Prompt #9: Data Processing Agent **Use case:** An agent that transforms, validates, and analyzes data. ``` You are a data processing agent. You handle data transformations, validation, and analysis with precision. ## Data Handling Principles 1. VALIDATE inputs before processing 2. PRESERVE original data — never modify source data 3. TRANSFORM in clear, reproducible steps 4. VERIFY outputs against expected schemas/ranges 5. DOCUMENT what was done to the data ## Input Validation Checklist - [ ] Correct format (CSV, JSON, etc.)? - [ ] Expected columns/fields present? - [ ] Data types match expectations? - [ ] Null/missing values identified? - [ ] Duplicates checked? - [ ] Value ranges reasonable? ## Processing Standards - Log each transformation step - Handle missing data explicitly (drop, fill, flag — never ignore) - Use consistent date/number formats throughout - Preserve data lineage (track where each value came from) ## Output Requirements Always include: 1. Summary statistics (count, completeness, key metrics) 2. Data quality report (issues found, actions taken) 3. The processed data in the requested format 4. Any warnings or anomalies detected ## Error Handling - Malformed input → report specific issue with row/field reference - Type mismatch → attempt safe conversion, flag if ambiguous - Missing required field → halt and report (don't guess) - Outlier detected → flag but include (don't silently remove) ``` **Usage Notes:** Use for ETL pipelines, data cleaning bots, and analytics agents. The validation checklist ensures data quality issues are caught early. --- ### Prompt #10: API Integration Agent **Use case:** An agent that interacts with external APIs reliably. ``` You are an API integration agent. You make reliable, well-formed API requests and handle responses gracefully. ## Request Standards - Always validate request parameters before sending - Use appropriate HTTP methods (GET for reads, POST for creates, etc.) - Include required headers (Content-Type, Authorization, etc.) - Set reasonable timeouts (default: 30s) - Never hardcode credentials — use environment variables or secrets managers ## Response Handling For every API response: 1. Check HTTP status code first 2. Validate response body structure 3. Extract the data you need 4. Handle errors by status category: - 2xx: Success — proceed with data - 3xx: Redirect — follow if appropriate - 4xx: Client error — fix the request - 429: Rate limited — wait and retry with backoff - 5xx: Server error — retry with exponential backoff ## Rate Limiting - Track API call counts per endpoint - Implement exponential backoff: wait 1s, 2s, 4s, 8s, then fail - Respect Retry-After headers when present - Queue non-urgent requests during rate limit periods ## Authentication Patterns - Bearer tokens: Include in Authorization header - API keys: Include in header or query parameter as documented - OAuth: Refresh tokens before they expire - Never log or expose authentication credentials ## Logging Log (redacting sensitive data): - Request: method, URL, timestamp - Response: status code, timing, data size - Errors: full error context for debugging ``` **Usage Notes:** Essential for any agent that calls external services. Pair with the error handling patterns in Section 4 for production reliability. --- ### Prompt #11: File & Document Management Agent **Use case:** An agent that reads, creates, modifies, and organizes files. ``` You are a file management agent. You handle documents and files carefully and systematically. ## Core Rules 1. NEVER overwrite a file without confirming or creating a backup 2. Validate file paths before any operation 3. Check file existence before reading 4. Verify disk space before writing large files 5. Use appropriate encodings (UTF-8 by default) ## File Operations Workflow ### Reading 1. Verify the file exists 2. Check file size (warn if > 10MB) 3. Read with appropriate encoding 4. Validate content (not empty, expected format) ### Writing 1. Verify target directory exists 2. Check if file already exists (confirm overwrite) 3. Write to a temp file first 4. Validate the written content 5. Move temp file to final location (atomic write) ### Modifying 1. Read the original file 2. Create a backup (file.bak or timestamped) 3. Apply modifications 4. Validate the result 5. Write the modified version ## Naming Conventions - Use lowercase with hyphens: `my-document-v2.md` - Include version or date for iterative files - Use standard extensions (.md, .json, .csv, etc.) - Avoid spaces and special characters in filenames ## Output Format After each operation: - Operation: [read/write/modify/delete] - File: [path] - Status: [success/failure] - Details: [size, encoding, modifications made] ``` **Usage Notes:** Good for agents that automate document workflows, file organization, or content management. The atomic write pattern prevents data corruption. --- ### Prompt #12: Testing & Quality Assurance Agent **Use case:** An agent that validates code, tests functionality, and ensures quality. ``` You are a QA agent. Your job is to find bugs, validate correctness, and ensure quality. ## Testing Strategy For any piece of code or functionality: 1. UNDERSTAND: What should it do? What are the requirements? 2. HAPPY PATH: Test the normal, expected use case 3. EDGE CASES: Test boundary conditions and unusual inputs 4. ERROR CASES: Test what happens when things go wrong 5. REGRESSION: Check that existing functionality still works ## Test Categories ### Unit Tests - Test individual functions in isolation - Mock external dependencies - Cover: normal input, empty input, null/undefined, type mismatches ### Integration Tests - Test components working together - Use real (or realistic) dependencies - Cover: data flow between components, error propagation ### Validation Tests - Input validation (type, range, format) - Output validation (schema, completeness) - State validation (before/after operations) ## Bug Report Format ``` **Bug:** [Brief description] **Severity:** Critical / High / Medium / Low **Steps to Reproduce:** 1. [Step 1] 2. [Step 2] **Expected:** [What should happen] **Actual:** [What actually happens] **Root Cause:** [If identified] **Suggested Fix:** [If known] ``` ## Quality Checklist - [ ] All happy paths work correctly - [ ] Edge cases handled (empty, null, max values) - [ ] Error messages are clear and actionable - [ ] No hardcoded values that should be configurable - [ ] Performance is acceptable for expected data sizes - [ ] Security: no injection vulnerabilities, inputs sanitized ``` **Usage Notes:** Deploy as a code review bot, CI/CD quality gate, or testing automation agent. The structured bug report format integrates well with issue trackers. --- ### Prompt #13: Memory & Context Management Agent **Use case:** An agent that maintains conversation context and long-term memory. ``` You are a context-aware agent with memory management capabilities. ## Memory Architecture ### Working Memory (Current Session) - Track the current task and its status - Maintain a list of decisions made and their reasons - Keep a running summary of the conversation ### Short-Term Memory (Recent Context) - Last 5-10 interactions summarized - Recently accessed files/resources - Pending follow-up items ### Long-Term Memory (Persistent) - User preferences and patterns - Project-specific knowledge - Frequently used configurations - Past solutions to similar problems ## Context Window Management When the context is getting long: 1. Summarize older interactions (keep decisions, discard details) 2. Reference stored memories instead of repeating information 3. Prioritize: current task > recent context > historical context 4. Archive completed task details to long-term memory ## Memory Operations - STORE: Save important information with a key and timestamp - RECALL: Retrieve relevant memories for the current task - UPDATE: Modify existing memories with new information - FORGET: Remove outdated or irrelevant memories - SUMMARIZE: Compress detailed memories into summaries ## Context Handoff When resuming or handing off: 1. Provide a 2-sentence situation summary 2. List active tasks and their status 3. Note any pending decisions or blockers 4. Reference relevant stored memories ``` **Usage Notes:** Essential for long-running agents and multi-session workflows. Implement with a vector database or key-value store for the long-term memory layer. --- ### Prompt #14: Autonomous Decision-Making Agent **Use case:** An agent that can make independent decisions within defined boundaries. ``` You are an autonomous agent authorized to make decisions within defined boundaries. ## Decision Authority Levels ### Level 1: Act Independently - Formatting and organizing outputs - Choosing between equivalent approaches - Retrying failed operations - Gathering additional information ### Level 2: Act and Notify - Making non-destructive changes to files - Calling external APIs for read operations - Creating drafts or temporary files - Selecting tools from the available set ### Level 3: Propose and Wait for Approval - Modifying existing data or configurations - Making external write operations (POST, PUT, DELETE) - Spending budget or resources - Actions that affect other users ### Level 4: Never Do (Escalate Immediately) - Deleting production data - Changing security settings - Making financial commitments - Actions outside your defined scope ## Decision Framework For each decision: 1. Classify the decision level (1-4) 2. If Level 1-2: proceed 3. If Level 3: present options with pros/cons, recommend one, wait 4. If Level 4: escalate immediately with context ## Logging Decisions Every decision should be logged: - What was decided - Why (brief justification) - What alternatives were considered - Authority level used ``` **Usage Notes:** Critical for production agents. Adjust the four levels based on your risk tolerance. Pair with the Safety prompt (#6) for defense-in-depth. --- ### Prompt #15: Workflow Orchestration Agent **Use case:** An agent that coordinates multiple sub-agents or tasks. ``` You are a workflow orchestration agent that coordinates complex multi-step processes. ## Orchestration Principles 1. Each workflow has a clear start, execution path, and end state 2. Sub-tasks should be independent when possible (parallelize) 3. Dependencies must be explicit 4. Every workflow must have a failure/rollback path ## Workflow Definition ``` Workflow: [Name] Trigger: [What starts this workflow] Steps: 1. [Task] → assigns to [agent/tool] → output: [expected] 2. [Task] → depends on: step 1 → output: [expected] ... Success Criteria: [How to know it worked] Failure Handling: [What to do if it fails] ``` ## Execution Engine - Run independent steps in parallel - Wait for dependencies before proceeding - Set per-step timeouts - Capture outputs from each step for downstream use ## Monitoring Track for each workflow execution: - Start time and estimated completion - Current step and overall progress (X/N steps) - Any warnings or partial failures - Resource usage ## Error Escalation 1. Step fails → retry (max 2 attempts) 2. Retry fails → try fallback approach 3. Fallback fails → pause workflow, notify orchestrator 4. Orchestrator can: skip step, substitute, or abort workflow ## Completion On workflow completion: - Verify all success criteria are met - Compile results from all steps - Clean up temporary resources - Log total execution time and any issues ``` **Usage Notes:** Use for agents that manage pipelines (CI/CD, data processing, content generation). The parallel execution model is key for performance. --- ### Prompt #16: Debugging & Troubleshooting Agent **Use case:** An agent that systematically diagnoses and resolves issues. ``` You are a debugging agent. You systematically diagnose and resolve issues. ## Debugging Protocol 1. REPRODUCE: Can you consistently reproduce the issue? 2. ISOLATE: What's the smallest input/scenario that triggers it? 3. HYPOTHESIZE: What could cause this behavior? (list 3+ hypotheses) 4. TEST: Design a test for each hypothesis 5. IDENTIFY: Confirm the root cause 6. FIX: Implement and verify the fix 7. PREVENT: Suggest how to prevent recurrence ## Information Gathering Start by collecting: - Error messages (exact text) - Stack traces - Environment details (OS, runtime version, dependencies) - Recent changes that might be related - Whether it worked before (and what changed) ## Hypothesis Ranking Rank hypotheses by: 1. Most likely based on symptoms 2. Easiest to test first 3. Most impactful if true ## Common Root Causes (Check These First) - Configuration mismatch (dev vs prod) - Missing or wrong environment variables - Dependency version conflicts - Race conditions / timing issues - Null/undefined values propagating - Resource exhaustion (memory, connections, disk) - Permission errors ## Fix Verification After applying a fix: 1. Reproduce the original scenario → should pass now 2. Run related tests → no regressions 3. Test edge cases near the fix → still stable 4. Document what was changed and why ``` **Usage Notes:** Excellent for incident response bots and development assistance agents. The structured hypothesis approach prevents random guessing. --- ### Prompt #17: Content Generation Agent **Use case:** An agent that creates written content with consistent quality. ``` You are a content generation agent that produces clear, engaging, well-structured writing. ## Content Principles 1. CLARITY: Every sentence should be easily understood on first read 2. STRUCTURE: Use headers, lists, and sections for scannability 3. VALUE: Every paragraph should teach, inform, or help the reader 4. ACCURACY: Never state something you're not confident about 5. VOICE: Match the requested tone (formal, casual, technical, etc.) ## Writing Workflow 1. Understand the brief (audience, purpose, tone, length) 2. Create an outline 3. Write the first draft (focus on content, not perfection) 4. Edit for clarity and conciseness 5. Format for the target medium (blog, email, docs, etc.) ## Quality Standards - No filler phrases ("In today's world...", "It's worth noting that...") - Active voice preferred over passive - Specific examples over vague generalizations - Short paragraphs (3-5 sentences max) - Consistent terminology throughout ## Output Formats Supported formats: - **Blog post**: Title, intro hook, sections with headers, conclusion with CTA - **Documentation**: Overview, prerequisites, step-by-step, examples, troubleshooting - **Email**: Subject line, greeting, body (3 paragraphs max), CTA, sign-off - **Social post**: Hook, value, CTA (platform character limits respected) ## Review Checklist - [ ] Matches the requested tone and audience - [ ] No factual errors or unsupported claims - [ ] Headers and formatting are consistent - [ ] Conclusion includes a clear next step - [ ] Length is within the requested range ``` **Usage Notes:** Use for blog writing, documentation, email drafting, and social media. The Review Checklist ensures consistent quality. --- ## Tool-Calling Templates ### Template 1: Basic Function Definition (OpenAI Format) ```json { "type": "function", "function": { "name": "search_database", "description": "Search the product database by query string. Returns matching products with name, price, and availability. Use this when the user asks about product information, pricing, or stock.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query. Can be a product name, category, or keyword." }, "max_results": { "type": "integer", "description": "Maximum number of results to return (1-50). Default: 10.", "default": 10 }, "sort_by": { "type": "string", "enum": ["relevance", "price_asc", "price_desc", "newest"], "description": "Sort order for results. Default: relevance." }, "in_stock_only": { "type": "boolean", "description": "If true, only return products currently in stock.", "default": false } }, "required": ["query"] } } } ``` **Usage Notes:** Follow this pattern for all function definitions. Key principles: - `name`: verb_noun format, lowercase with underscores - `description`: What it does + when to use it - `parameters`: Include descriptions for every parameter - `required`: Only truly required params --- ### Template 2: Anthropic Tool Format ```json { "name": "execute_code", "description": "Execute a code snippet in a sandboxed environment. Returns stdout, stderr, and exit code. Supports Python, JavaScript, and Bash. Use this when the user wants to run code, test a function, or verify output.", "input_schema": { "type": "object", "properties": { "language": { "type": "string", "enum": ["python", "javascript", "bash"], "description": "The programming language of the code to execute." }, "code": { "type": "string", "description": "The code to execute. Must be a complete, runnable snippet." }, "timeout_seconds": { "type": "integer", "description": "Maximum execution time in seconds (1-60). Default: 10.", "default": 10 }, "environment_vars": { "type": "object", "description": "Key-value pairs of environment variables to set before execution.", "additionalProperties": { "type": "string" } } }, "required": ["language", "code"] } } ``` --- ### Template 3: Multi-Tool Router ```python # Tool Router — routes requests to the appropriate tool based on intent TOOL_REGISTRY = { "search": { "description": "Search for information", "handler": search_handler, "requires_auth": False, "rate_limit": 10, # per minute }, "write_file": { "description": "Create or modify a file", "handler": file_write_handler, "requires_auth": True, "rate_limit": 5, }, "send_email": { "description": "Send an email", "handler": email_handler, "requires_auth": True, "rate_limit": 3, }, "query_database": { "description": "Run a database query", "handler": db_handler, "requires_auth": True, "rate_limit": 20, }, } class ToolRouter: def __init__(self, registry: dict): self.registry = registry self.call_counts = {} # rate limiting tracker def route(self, tool_name: str, params: dict, auth_context=None) -> dict: """Route a tool call to the appropriate handler.""" # 1. Validate tool exists if tool_name not in self.registry: return {"error": f"Unknown tool: {tool_name}", "available": list(self.registry.keys())} tool = self.registry[tool_name] # 2. Check auth if required if tool["requires_auth"] and not auth_context: return {"error": f"Tool '{tool_name}' requires authentication"} # 3. Check rate limit if not self._check_rate_limit(tool_name, tool["rate_limit"]): return {"error": f"Rate limit exceeded for '{tool_name}'", "retry_after_seconds": 60} # 4. Execute try: result = tool["handler"](params, auth_context) return {"success": True, "data": result} except Exception as e: return {"error": str(e), "tool": tool_name} def _check_rate_limit(self, tool_name: str, limit: int) -> bool: # Simplified — use a proper rate limiter in production count = self.call_counts.get(tool_name, 0) if count >= limit: return False self.call_counts[tool_name] = count + 1 return True ``` --- ### Template 4: Tool Result Parser ```python class ToolResultParser: """Parse and validate tool call results before passing to the LLM.""" @staticmethod def parse(tool_name: str, raw_result: dict) -> dict: """ Parse a raw tool result into a standardized format. Returns: { "status": "success" | "error" | "partial", "data": , "metadata": {"tool": str, "timestamp": str, "duration_ms": int} } """ parsed = { "status": "success", "data": None, "metadata": { "tool": tool_name, "timestamp": datetime.utcnow().isoformat(), } } # Check for errors if "error" in raw_result: parsed["status"] = "error" parsed["data"] = { "error_message": raw_result["error"], "error_code": raw_result.get("code", "UNKNOWN"), "recoverable": raw_result.get("recoverable", True), } return parsed # Check for partial results if raw_result.get("truncated") or raw_result.get("partial"): parsed["status"] = "partial" parsed["data"] = raw_result.get("data", raw_result) parsed["metadata"]["warning"] = "Results may be incomplete" return parsed # Success parsed["data"] = raw_result.get("data", raw_result) return parsed @staticmethod def summarize_for_context(parsed_result: dict, max_length: int = 500) -> str: """Create a concise summary of a tool result for the LLM context.""" if parsed_result["status"] == "error": return f"[Tool Error] {parsed_result['data']['error_message']}" data = parsed_result["data"] summary = json.dumps(data, indent=2) if len(summary) > max_length: summary = summary[:max_length] + "\n... [truncated]" prefix = "[Tool Result]" if parsed_result["status"] == "success" else "[Partial Result]" return f"{prefix}\n{summary}" ``` --- ### Template 5: Parallel Tool Execution ```python import asyncio from typing import List, Dict, Any class ParallelToolExecutor: """Execute multiple independent tool calls in parallel.""" def __init__(self, tool_router: ToolRouter, max_concurrent: int = 5): self.router = tool_router self.semaphore = asyncio.Semaphore(max_concurrent) async def execute_batch( self, tool_calls: List[Dict[str, Any]], auth_context=None ) -> List[Dict[str, Any]]: """ Execute multiple tool calls in parallel. Args: tool_calls: [{"tool": "name", "params": {...}}, ...] Returns: [{"tool": "name", "result": {...}, "duration_ms": int}, ...] """ tasks = [ self._execute_one(call, auth_context) for call in tool_calls ] return await asyncio.gather(*tasks) async def _execute_one( self, call: Dict[str, Any], auth_context ) -> Dict[str, Any]: async with self.semaphore: start = asyncio.get_event_loop().time() try: result = self.router.route( call["tool"], call["params"], auth_context ) duration = int((asyncio.get_event_loop().time() - start) * 1000) return { "tool": call["tool"], "result": result, "duration_ms": duration } except Exception as e: return { "tool": call["tool"], "result": {"error": str(e)}, "duration_ms": 0 } # Usage: # executor = ParallelToolExecutor(router) # results = await executor.execute_batch([ # {"tool": "search", "params": {"query": "AI agents"}}, # {"tool": "query_database", "params": {"sql": "SELECT count(*) FROM users"}}, # ]) ``` --- ## Error-Handling Patterns ### Pattern 1: Retry with Exponential Backoff ```python import time import random from typing import Callable, TypeVar, Optional T = TypeVar('T') def retry_with_backoff( fn: Callable[..., T], max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 30.0, jitter: bool = True, retryable_exceptions: tuple = (Exception,), ) -> T: """ Retry a function with exponential backoff. Args: fn: The function to retry max_retries: Maximum number of retry attempts base_delay: Initial delay in seconds max_delay: Maximum delay between retries jitter: Add randomness to prevent thundering herd retryable_exceptions: Only retry on these exception types """ last_exception = None for attempt in range(max_retries + 1): try: return fn() except retryable_exceptions as e: last_exception = e if attempt == max_retries: break # Calculate delay with exponential backoff delay = min(base_delay * (2 ** attempt), max_delay) # Add jitter (±25%) if jitter: delay = delay * (0.75 + random.random() * 0.5) print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s...") time.sleep(delay) raise last_exception # Usage: # result = retry_with_backoff( # lambda: api_client.get("/data"), # max_retries=3, # retryable_exceptions=(ConnectionError, TimeoutError) # ) ``` --- ### Pattern 2: Fallback Chain ```python from typing import List, Callable, Any, Optional from dataclasses import dataclass @dataclass class FallbackOption: name: str handler: Callable description: str class FallbackChain: """ Try multiple approaches in sequence until one succeeds. Example: Try primary API → Try backup API → Use cached data → Return default """ def __init__(self, options: List[FallbackOption]): self.options = options self.execution_log = [] def execute(self, *args, **kwargs) -> dict: """ Execute the fallback chain. Returns: { "success": bool, "data": Any, "used_fallback": str, # name of the option that succeeded "attempts": [{"name": str, "error": str | None}] } """ self.execution_log = [] for option in self.options: try: result = option.handler(*args, **kwargs) self.execution_log.append({"name": option.name, "error": None}) return { "success": True, "data": result, "used_fallback": option.name, "attempts": self.execution_log } except Exception as e: self.execution_log.append({"name": option.name, "error": str(e)}) continue return { "success": False, "data": None, "used_fallback": None, "attempts": self.execution_log } # Usage: # chain = FallbackChain([ # FallbackOption("primary_api", call_primary_api, "Main API endpoint"), # FallbackOption("backup_api", call_backup_api, "Backup API endpoint"), # FallbackOption("cache", read_from_cache, "Local cache lookup"), # FallbackOption("default", lambda: DEFAULT_VALUE, "Hardcoded default"), # ]) # result = chain.execute(query="search term") ``` --- ### Pattern 3: Graceful Degradation ```python from enum import Enum from typing import Dict, Any, Optional from dataclasses import dataclass, field class ServiceHealth(Enum): HEALTHY = "healthy" DEGRADED = "degraded" DOWN = "down" @dataclass class ServiceStatus: name: str health: ServiceHealth last_check: float error_count: int = 0 last_error: Optional[str] = None class GracefulDegradation: """ Monitor service health and automatically degrade functionality when dependencies are unavailable. """ def __init__(self, error_threshold: int = 3): self.services: Dict[str, ServiceStatus] = {} self.error_threshold = error_threshold def register_service(self, name: str): self.services[name] = ServiceStatus( name=name, health=ServiceHealth.HEALTHY, last_check=time.time() ) def record_success(self, service_name: str): svc = self.services[service_name] svc.health = ServiceHealth.HEALTHY svc.error_count = 0 svc.last_check = time.time() def record_failure(self, service_name: str, error: str): svc = self.services[service_name] svc.error_count += 1 svc.last_error = error svc.last_check = time.time() if svc.error_count >= self.error_threshold: svc.health = ServiceHealth.DOWN elif svc.error_count >= self.error_threshold // 2: svc.health = ServiceHealth.DEGRADED def get_available_features(self) -> Dict[str, bool]: """Return which features are currently available.""" return { name: status.health != ServiceHealth.DOWN for name, status in self.services.items() } def get_status_report(self) -> str: lines = ["Service Health Report:"] for name, status in self.services.items(): icon = {"healthy": "✓", "degraded": "⚠", "down": "✗"}[status.health.value] lines.append(f" {icon} {name}: {status.health.value}") if status.last_error: lines.append(f" Last error: {status.last_error}") return "\n".join(lines) # Usage: # degradation = GracefulDegradation(error_threshold=3) # degradation.register_service("search_api") # degradation.register_service("database") # degradation.register_service("cache") # # try: # result = search_api.search(query) # degradation.record_success("search_api") # except Exception as e: # degradation.record_failure("search_api", str(e)) # # Fall back to database search ``` --- ### Pattern 4: Circuit Breaker ```python import time from enum import Enum from threading import Lock class CircuitState(Enum): CLOSED = "closed" # Normal operation OPEN = "open" # Failing — reject requests immediately HALF_OPEN = "half_open" # Testing if service recovered class CircuitBreaker: """ Prevent cascading failures by stopping requests to a failing service. States: - CLOSED: Normal. Requests pass through. Track failures. - OPEN: Service is down. Reject requests immediately. Wait for reset timeout. - HALF_OPEN: Allow one test request. If it succeeds → CLOSED. If it fails → OPEN. """ def __init__( self, failure_threshold: int = 5, reset_timeout: float = 30.0, name: str = "default" ): self.failure_threshold = failure_threshold self.reset_timeout = reset_timeout self.name = name self.state = CircuitState.CLOSED self.failure_count = 0 self.last_failure_time = 0 self.lock = Lock() def call(self, fn, *args, **kwargs): """Execute a function through the circuit breaker.""" with self.lock: if self.state == CircuitState.OPEN: if time.time() - self.last_failure_time >= self.reset_timeout: self.state = CircuitState.HALF_OPEN else: raise CircuitBreakerOpen( f"Circuit '{self.name}' is OPEN. " f"Retry after {self.reset_timeout}s." ) try: result = fn(*args, **kwargs) self._record_success() return result except Exception as e: self._record_failure() raise def _record_success(self): with self.lock: self.failure_count = 0 self.state = CircuitState.CLOSED def _record_failure(self): with self.lock: self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = CircuitState.OPEN class CircuitBreakerOpen(Exception): pass # Usage: # breaker = CircuitBreaker(failure_threshold=5, reset_timeout=30, name="payment_api") # try: # result = breaker.call(payment_api.charge, amount=999) # except CircuitBreakerOpen: # # Service is down — use fallback # result = queue_for_later(amount=999) ``` --- ### Pattern 5: Structured Error Responses ```python from dataclasses import dataclass from typing import Optional, List, Dict, Any from enum import Enum class ErrorSeverity(Enum): LOW = "low" # Non-blocking, informational MEDIUM = "medium" # Degraded functionality HIGH = "high" # Feature unavailable CRITICAL = "critical" # System-level failure @dataclass class AgentError: """Standardized error format for AI agent systems.""" code: str # e.g., "TOOL_TIMEOUT", "AUTH_FAILED" message: str # Human-readable description severity: ErrorSeverity recoverable: bool # Can the agent retry or work around this? context: Dict[str, Any] # Additional debugging info suggestions: List[str] # What the agent (or user) can do next def to_dict(self) -> dict: return { "error": { "code": self.code, "message": self.message, "severity": self.severity.value, "recoverable": self.recoverable, "context": self.context, "suggestions": self.suggestions, } } def to_user_message(self) -> str: """Format for display to end users (no internal details).""" msg = self.message if self.suggestions: msg += "\n\nYou can try:\n" msg += "\n".join(f" • {s}" for s in self.suggestions) return msg # Pre-defined error templates ERRORS = { "TOOL_TIMEOUT": lambda tool, timeout: AgentError( code="TOOL_TIMEOUT", message=f"Tool '{tool}' did not respond within {timeout}s", severity=ErrorSeverity.MEDIUM, recoverable=True, context={"tool": tool, "timeout": timeout}, suggestions=["Retry the operation", "Try an alternative tool", "Reduce the request scope"] ), "RATE_LIMITED": lambda service: AgentError( code="RATE_LIMITED", message=f"Rate limit reached for '{service}'", severity=ErrorSeverity.LOW, recoverable=True, context={"service": service}, suggestions=["Wait and retry automatically", "Use cached results if available"] ), "AUTH_FAILED": lambda service: AgentError( code="AUTH_FAILED", message=f"Authentication failed for '{service}'", severity=ErrorSeverity.HIGH, recoverable=False, context={"service": service}, suggestions=["Check API credentials", "Refresh authentication token", "Contact support"] ), } ``` --- ## Integration Blueprints ### Blueprint 1: REST API Connector ```python import httpx from typing import Optional, Dict, Any from dataclasses import dataclass @dataclass class APIConfig: base_url: str api_key: Optional[str] = None timeout: float = 30.0 max_retries: int = 3 headers: Optional[Dict[str, str]] = None class RESTConnector: """ Production-ready REST API connector with retry, auth, and error handling. """ def __init__(self, config: APIConfig): self.config = config self.client = httpx.Client( base_url=config.base_url, timeout=config.timeout, headers=self._build_headers() ) def _build_headers(self) -> Dict[str, str]: headers = {"Content-Type": "application/json", "User-Agent": "AgentKit/1.0"} if self.config.api_key: headers["Authorization"] = f"Bearer {self.config.api_key}" if self.config.headers: headers.update(self.config.headers) return headers def get(self, path: str, params: Optional[Dict] = None) -> Dict[str, Any]: return self._request("GET", path, params=params) def post(self, path: str, data: Optional[Dict] = None) -> Dict[str, Any]: return self._request("POST", path, json=data) def put(self, path: str, data: Optional[Dict] = None) -> Dict[str, Any]: return self._request("PUT", path, json=data) def delete(self, path: str) -> Dict[str, Any]: return self._request("DELETE", path) def _request(self, method: str, path: str, **kwargs) -> Dict[str, Any]: last_error = None for attempt in range(self.config.max_retries): try: response = self.client.request(method, path, **kwargs) if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 2 ** attempt)) time.sleep(retry_after) continue response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: if e.response.status_code < 500: raise # Client errors are not retryable last_error = e time.sleep(2 ** attempt) except (httpx.ConnectError, httpx.ReadTimeout) as e: last_error = e time.sleep(2 ** attempt) raise last_error # Usage: # api = RESTConnector(APIConfig( # base_url="https://api.example.com", # api_key=os.environ["API_KEY"], # timeout=15.0 # )) # users = api.get("/v1/users", params={"limit": 10}) ``` --- ### Blueprint 2: Webhook Handler ```python import hmac import hashlib import json from typing import Callable, Dict, Any, Optional from dataclasses import dataclass, field @dataclass class WebhookConfig: secret: str # For signature verification event_handlers: Dict[str, Callable] # event_type → handler function retry_on_failure: bool = True max_processing_time: float = 25.0 # seconds (leave margin for platform timeout) class WebhookHandler: """ Production webhook receiver with signature verification and event routing. """ def __init__(self, config: WebhookConfig): self.config = config self.processed_events = set() # Idempotency tracking def verify_signature(self, payload: bytes, signature: str) -> bool: """Verify webhook signature (HMAC-SHA256).""" expected = hmac.new( self.config.secret.encode(), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(f"sha256={expected}", signature) def handle(self, payload: bytes, signature: Optional[str] = None) -> Dict[str, Any]: """ Process an incoming webhook. Returns: {"status": "processed"|"skipped"|"error", "event_type": str, "details": str} """ # 1. Verify signature if provided if signature and not self.verify_signature(payload, signature): return {"status": "error", "details": "Invalid signature"} # 2. Parse payload try: event = json.loads(payload) except json.JSONDecodeError: return {"status": "error", "details": "Invalid JSON payload"} event_type = event.get("event_type", "unknown") event_id = event.get("id", "") # 3. Idempotency check if event_id and event_id in self.processed_events: return {"status": "skipped", "event_type": event_type, "details": "Already processed"} # 4. Route to handler handler = self.config.event_handlers.get(event_type) if not handler: return {"status": "skipped", "event_type": event_type, "details": "No handler registered"} # 5. Execute handler try: handler(event) if event_id: self.processed_events.add(event_id) return {"status": "processed", "event_type": event_type, "details": "Success"} except Exception as e: return {"status": "error", "event_type": event_type, "details": str(e)} # Usage: # def on_payment(event): # email = event["payment"]["customer_email"] # amount = event["payment"]["amount_cents"] / 100 # send_confirmation(email, amount) # # handler = WebhookHandler(WebhookConfig( # secret=os.environ["WEBHOOK_SECRET"], # event_handlers={ # "checkout.session.completed": on_payment, # "subscription.created": on_subscription, # } # )) ``` --- ### Blueprint 3: State Machine for Agent Workflows ```python from enum import Enum from typing import Dict, Callable, Any, Optional, List from dataclasses import dataclass, field class AgentState(Enum): IDLE = "idle" PLANNING = "planning" EXECUTING = "executing" WAITING_FOR_INPUT = "waiting_for_input" WAITING_FOR_TOOL = "waiting_for_tool" ERROR = "error" COMPLETED = "completed" @dataclass class Transition: from_state: AgentState to_state: AgentState trigger: str guard: Optional[Callable] = None # Condition that must be true action: Optional[Callable] = None # Side effect on transition class AgentStateMachine: """ Manage agent lifecycle with explicit states and transitions. Prevents invalid state changes and enables workflow monitoring. """ def __init__(self, initial_state: AgentState = AgentState.IDLE): self.state = initial_state self.transitions: Dict[str, List[Transition]] = {} self.history: List[Dict[str, Any]] = [] self.context: Dict[str, Any] = {} self._setup_default_transitions() def _setup_default_transitions(self): defaults = [ Transition(AgentState.IDLE, AgentState.PLANNING, "start_task"), Transition(AgentState.PLANNING, AgentState.EXECUTING, "plan_ready"), Transition(AgentState.EXECUTING, AgentState.WAITING_FOR_TOOL, "tool_called"), Transition(AgentState.WAITING_FOR_TOOL, AgentState.EXECUTING, "tool_returned"), Transition(AgentState.EXECUTING, AgentState.WAITING_FOR_INPUT, "need_input"), Transition(AgentState.WAITING_FOR_INPUT, AgentState.EXECUTING, "input_received"), Transition(AgentState.EXECUTING, AgentState.COMPLETED, "task_done"), Transition(AgentState.EXECUTING, AgentState.ERROR, "error_occurred"), Transition(AgentState.ERROR, AgentState.EXECUTING, "error_recovered"), Transition(AgentState.ERROR, AgentState.COMPLETED, "abort"), ] for t in defaults: self.add_transition(t) def add_transition(self, transition: Transition): key = f"{transition.from_state.value}:{transition.trigger}" if key not in self.transitions: self.transitions[key] = [] self.transitions[key].append(transition) def trigger(self, event: str, **kwargs) -> bool: key = f"{self.state.value}:{event}" transitions = self.transitions.get(key, []) for t in transitions: if t.guard and not t.guard(self.context, kwargs): continue # Execute transition old_state = self.state if t.action: t.action(self.context, kwargs) self.state = t.to_state self.history.append({ "from": old_state.value, "to": self.state.value, "trigger": event, "timestamp": time.time() }) return True return False # No valid transition found def get_status(self) -> Dict[str, Any]: return { "current_state": self.state.value, "transitions_count": len(self.history), "context": self.context, "available_triggers": self._get_available_triggers() } def _get_available_triggers(self) -> List[str]: triggers = set() for key in self.transitions: state, trigger = key.split(":") if state == self.state.value: triggers.add(trigger) return sorted(triggers) # Usage: # sm = AgentStateMachine() # sm.trigger("start_task") # IDLE → PLANNING # sm.trigger("plan_ready") # PLANNING → EXECUTING # sm.trigger("tool_called") # EXECUTING → WAITING_FOR_TOOL # sm.trigger("tool_returned") # WAITING_FOR_TOOL → EXECUTING # sm.trigger("task_done") # EXECUTING → COMPLETED # print(sm.get_status()) ``` --- ### Blueprint 4: Event Bus for Agent Communication ```python import asyncio from typing import Callable, Dict, List, Any from dataclasses import dataclass from collections import defaultdict @dataclass class Event: type: str source: str data: Dict[str, Any] timestamp: float = None def __post_init__(self): if self.timestamp is None: self.timestamp = time.time() class EventBus: """ Lightweight event bus for agent-to-agent and agent-to-system communication. Supports both sync and async handlers. """ def __init__(self): self.handlers: Dict[str, List[Callable]] = defaultdict(list) self.event_log: List[Event] = [] self.max_log_size = 1000 def subscribe(self, event_type: str, handler: Callable): """Register a handler for an event type. Use '*' for all events.""" self.handlers[event_type].append(handler) def unsubscribe(self, event_type: str, handler: Callable): self.handlers[event_type].remove(handler) def publish(self, event: Event): """Publish an event to all registered handlers.""" self._log_event(event) # Specific handlers for handler in self.handlers.get(event.type, []): try: handler(event) except Exception as e: print(f"Handler error for {event.type}: {e}") # Wildcard handlers for handler in self.handlers.get("*", []): try: handler(event) except Exception as e: print(f"Wildcard handler error: {e}") async def publish_async(self, event: Event): """Async version of publish.""" self._log_event(event) tasks = [] for handler in self.handlers.get(event.type, []) + self.handlers.get("*", []): if asyncio.iscoroutinefunction(handler): tasks.append(handler(event)) else: handler(event) if tasks: await asyncio.gather(*tasks, return_exceptions=True) def _log_event(self, event: Event): self.event_log.append(event) if len(self.event_log) > self.max_log_size: self.event_log = self.event_log[-self.max_log_size:] def get_recent_events(self, event_type: str = None, limit: int = 10) -> List[Event]: events = self.event_log if event_type: events = [e for e in events if e.type == event_type] return events[-limit:] # Usage: # bus = EventBus() # # # Subscribe handlers # bus.subscribe("task.completed", lambda e: print(f"Task done: {e.data}")) # bus.subscribe("error.occurred", lambda e: alert_team(e.data)) # bus.subscribe("*", lambda e: log_to_db(e)) # Log everything # # # Publish events # bus.publish(Event(type="task.completed", source="agent-1", data={"task_id": "abc"})) ``` --- ### Blueprint 5: Agent Memory Store ```python import json import time from typing import Dict, Any, Optional, List from dataclasses import dataclass @dataclass class Memory: key: str value: Any category: str # "preference", "fact", "task", "context" created_at: float updated_at: float access_count: int = 0 ttl: Optional[float] = None # Time-to-live in seconds class AgentMemoryStore: """ Simple key-value memory store for agent context management. Supports categories, TTL, and access-frequency tracking. """ def __init__(self, max_memories: int = 500): self.memories: Dict[str, Memory] = {} self.max_memories = max_memories def store(self, key: str, value: Any, category: str = "general", ttl: float = None): now = time.time() self.memories[key] = Memory( key=key, value=value, category=category, created_at=now, updated_at=now, ttl=ttl ) self._evict_if_needed() def recall(self, key: str) -> Optional[Any]: mem = self.memories.get(key) if not mem: return None if mem.ttl and (time.time() - mem.updated_at) > mem.ttl: del self.memories[key] return None mem.access_count += 1 return mem.value def search(self, category: str = None, keyword: str = None) -> List[Memory]: results = list(self.memories.values()) if category: results = [m for m in results if m.category == category] if keyword: keyword_lower = keyword.lower() results = [ m for m in results if keyword_lower in str(m.value).lower() or keyword_lower in m.key.lower() ] return sorted(results, key=lambda m: m.access_count, reverse=True) def forget(self, key: str) -> bool: if key in self.memories: del self.memories[key] return True return False def summarize(self) -> Dict[str, Any]: categories = {} for mem in self.memories.values(): if mem.category not in categories: categories[mem.category] = 0 categories[mem.category] += 1 return { "total_memories": len(self.memories), "categories": categories, "most_accessed": sorted( self.memories.values(), key=lambda m: m.access_count, reverse=True )[:5] } def _evict_if_needed(self): if len(self.memories) <= self.max_memories: return # Evict least recently accessed memories sorted_mems = sorted(self.memories.values(), key=lambda m: m.access_count) to_remove = len(self.memories) - self.max_memories for mem in sorted_mems[:to_remove]: del self.memories[mem.key] # Usage: # memory = AgentMemoryStore(max_memories=200) # memory.store("user_language", "Python", category="preference") # memory.store("project_framework", "FastAPI", category="context") # memory.store("last_error", "Connection timeout", category="debug", ttl=3600) # # lang = memory.recall("user_language") # "Python" # prefs = memory.search(category="preference") ``` --- ## Appendix ### Prompt Composition Pattern Combine prompts from this kit by layering: ``` SYSTEM_PROMPT = f""" {SAFETY_PROMPT} {ROLE_PROMPT} {TOOL_PROMPT} ## Available Tools {format_tools(tools)} """ ``` Layer order (top = highest priority): 1. Safety & Guardrails (Prompt #6) 2. Role-specific prompt (Prompts #1-5, #7-17) 3. Tool-using prompt (Prompt #5) 4. Memory management (Prompt #13) ### Compatibility All templates in this kit work with: - OpenAI GPT-4 / GPT-4o - Anthropic Claude 3.5 / Claude 3 Opus - Google Gemini - Mistral Large - Llama 3 70B+ - Any model supporting function calling ### Getting Help Questions? Issues? Email: support@survive.nanocorp.app --- *Built by SURVIVE. Made to work at any scale.*