Building Agents Patterns

Building Agents - Patterns & Best Practices

Design patterns, examples, and best practices for building robust goal-driven agents.

Prerequisites: Complete agent structure using building-agents-construction.

Practical Example: Hybrid Workflow

How to build a node using both direct file writes and optional MCP validation:

# 1. WRITE TO FILE FIRST (Primary - makes it visible)
node_code = '''
search_node = NodeSpec(
    id="search-web",
    node_type="llm_tool_use",
    input_keys=["query"],
    output_keys=["search_results"],
    system_prompt="Search the web for: {query}",
    tools=["web_search"],
)
'''

Edit(
    file_path="exports/research_agent/nodes/__init__.py",
    old_string="# Nodes will be added here",
    new_string=node_code
)

print("✅ Added search_node to nodes/__init__.py")
print("📁 Open exports/research_agent/nodes/__init__.py to see it!")

# 2. OPTIONALLY VALIDATE WITH MCP (Secondary - bookkeeping)
validation = mcp__agent-builder__test_node(
    node_id="search-web",
    test_input='{"query": "python tutorials"}',
    mock_llm_response='{"search_results": [...mock results...]}'
)

print(f"✓ Validation: {validation['success']}")

User experience:

• Immediately sees node in their editor (from step 1)
• Gets validation feedback (from step 2)
• Can edit the file directly if needed

This combines visibility (files) with validation (MCP tools).

Pause/Resume Architecture

For agents needing multi-turn conversations with user interaction:

Basic Pause/Resume Flow

# Define pause nodes - execution stops at these nodes
pause_nodes = ["request-clarification", "await-approval"]

# Define entry points - where to resume from each pause
entry_points = {
    "start": "analyze-request",  # Initial entry
    "request-clarification_resume": "process-clarification",  # Resume from clarification
    "await-approval_resume": "execute-action",  # Resume from approval
}

Example: Multi-Turn Research Agent

# Nodes
nodes = [
    NodeSpec(id="analyze-request", ...),
    NodeSpec(id="request-clarification", ...),  # PAUSE NODE
    NodeSpec(id="process-clarification", ...),
    NodeSpec(id="generate-results", ...),
    NodeSpec(id="await-approval", ...),  # PAUSE NODE
    NodeSpec(id="execute-action", ...),
]

# Edges with resume flows
edges = [
    EdgeSpec(
        id="analyze-to-clarify",
        source="analyze-request",
        target="request-clarification",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="needs_clarification == true",
    ),
    # When resumed, goes to process-clarification
    EdgeSpec(
        id="clarify-to-process",
        source="request-clarification",
        target="process-clarification",
        condition=EdgeCondition.ALWAYS,
    ),
    EdgeSpec(
        id="results-to-approval",
        source="generate-results",
        target="await-approval",
        condition=EdgeCondition.ALWAYS,
    ),
    # When resumed, goes to execute-action
    EdgeSpec(
        id="approval-to-execute",
        source="await-approval",
        target="execute-action",
        condition=EdgeCondition.ALWAYS,
    ),
]

# Configuration
pause_nodes = ["request-clarification", "await-approval"]
entry_points = {
    "start": "analyze-request",
    "request-clarification_resume": "process-clarification",
    "await-approval_resume": "execute-action",
}

Running Pause/Resume Agents

# Initial run - will pause at first pause node
result1 = await agent.run(
    context={"query": "research topic"},
    session_state=None
)

# Check if paused
if result1.paused_at:
    print(f"Paused at: {result1.paused_at}")

    # Resume with user input
    result2 = await agent.run(
        context={"user_response": "clarification details"},
        session_state=result1.session_state  # Pass previous state
    )

Anti-Patterns

What NOT to Do

❌ Don't rely on export_graph - Write files immediately, not at end

# BAD: Building in session state, exporting at end
mcp__agent-builder__add_node(...)
mcp__agent-builder__add_node(...)
mcp__agent-builder__export_graph()  # Files appear only now

# GOOD: Writing files immediately
Write(file_path="...", content=node_code)  # File visible now
Write(file_path="...", content=node_code)  # File visible now

❌ Don't hide code in session - Write to files as components approved

# BAD: Accumulating changes invisibly
session.add_component(component1)
session.add_component(component2)
# User can't see anything yet

# GOOD: Incremental visibility
Edit(file_path="...", ...)  # User sees change 1
Edit(file_path="...", ...)  # User sees change 2

❌ Don't wait to write files - Agent visible from first step

# BAD: Building everything before writing
design_all_nodes()
design_all_edges()
write_everything_at_once()

# GOOD: Write as you go
write_package_structure()  # Visible
write_goal()  # Visible
write_node_1()  # Visible
write_node_2()  # Visible

❌ Don't batch everything - Write incrementally

# BAD: Batching all nodes
nodes = [design_node_1(), design_node_2(), ...]
write_all_nodes(nodes)

# GOOD: One at a time with user feedback
write_node_1()  # User approves
write_node_2()  # User approves
write_node_3()  # User approves

MCP Tools - Correct Usage

MCP tools OK for: ✅ test_node - Validate node configuration with mock inputs ✅ validate_graph - Check graph structure ✅ create_session - Track session state for bookkeeping ✅ Other validation tools

Just don't: Use MCP as the primary construction method or rely on export_graph

Best Practices

1. Show Progress After Each Write

# After writing a node
print("✅ Added analyze_request_node to nodes/__init__.py")
print("📊 Progress: 1/6 nodes added")
print("📁 Open exports/my_agent/nodes/__init__.py to see it!")

2. Let User Open Files During Build

# Encourage file inspection
print("✅ Goal written to agent.py")
print("")
print("💡 Tip: Open exports/my_agent/agent.py in your editor to see the goal!")

3. Write Incrementally - One Component at a Time

# Good flow
write_package_structure()
show_user("Package created")

write_goal()
show_user("Goal written")

for node in nodes:
    get_approval(node)
    write_node(node)
    show_user(f"Node {node.id} written")

4. Test As You Build

# After adding several nodes
print("💡 You can test current state with:")
print("  PYTHONPATH=core:exports python -m my_agent validate")
print("  PYTHONPATH=core:exports python -m my_agent info")

5. Keep User Informed

# Clear status updates
print("🔨 Creating package structure...")
print("✅ Package created: exports/my_agent/")
print("")
print("📝 Next: Define agent goal")

Continuous Monitoring Agents

For agents that run continuously without terminal nodes:

# No terminal nodes - loops forever
terminal_nodes = []

# Workflow loops back to start
edges = [
    EdgeSpec(id="monitor-to-check", source="monitor", target="check-condition"),
    EdgeSpec(id="check-to-wait", source="check-condition", target="wait"),
    EdgeSpec(id="wait-to-monitor", source="wait", target="monitor"),  # Loop
]

# Entry node only
entry_node = "monitor"
entry_points = {"start": "monitor"}
pause_nodes = []

Example: File Monitor

nodes = [
    NodeSpec(id="list-files", ...),
    NodeSpec(id="check-new-files", node_type="router", ...),
    NodeSpec(id="process-files", ...),
    NodeSpec(id="wait-interval", node_type="function", ...),
]

edges = [
    EdgeSpec(id="list-to-check", source="list-files", target="check-new-files"),
    EdgeSpec(
        id="check-to-process",
        source="check-new-files",
        target="process-files",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="new_files_count > 0",
    ),
    EdgeSpec(
        id="check-to-wait",
        source="check-new-files",
        target="wait-interval",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="new_files_count == 0",
    ),
    EdgeSpec(id="process-to-wait", source="process-files", target="wait-interval"),
    EdgeSpec(id="wait-to-list", source="wait-interval", target="list-files"),  # Loop back
]

terminal_nodes = []  # No terminal - runs forever

Complex Routing Patterns

Multi-Condition Router

router_node = NodeSpec(
    id="decision-router",
    node_type="router",
    input_keys=["analysis_result"],
    output_keys=["decision"],
    system_prompt="""
    Based on the analysis result, decide the next action:
    - If confidence > 0.9: route to "execute"
    - If 0.5 <= confidence <= 0.9: route to "review"
    - If confidence < 0.5: route to "clarify"

    Return: {"decision": "execute|review|clarify"}
    """,
)

# Edges for each route
edges = [
    EdgeSpec(
        id="router-to-execute",
        source="decision-router",
        target="execute-action",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="decision == 'execute'",
        priority=1,
    ),
    EdgeSpec(
        id="router-to-review",
        source="decision-router",
        target="human-review",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="decision == 'review'",
        priority=2,
    ),
    EdgeSpec(
        id="router-to-clarify",
        source="decision-router",
        target="request-clarification",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="decision == 'clarify'",
        priority=3,
    ),
]

Error Handling Patterns

Graceful Failure with Fallback

# Primary node with error handling
nodes = [
    NodeSpec(id="api-call", max_retries=3, ...),
    NodeSpec(id="fallback-cache", ...),
    NodeSpec(id="report-error", ...),
]

edges = [
    # Success path
    EdgeSpec(
        id="api-success",
        source="api-call",
        target="process-results",
        condition=EdgeCondition.ON_SUCCESS,
    ),
    # Fallback on failure
    EdgeSpec(
        id="api-to-fallback",
        source="api-call",
        target="fallback-cache",
        condition=EdgeCondition.ON_FAILURE,
        priority=1,
    ),
    # Report if fallback also fails
    EdgeSpec(
        id="fallback-to-error",
        source="fallback-cache",
        target="report-error",
        condition=EdgeCondition.ON_FAILURE,
        priority=1,
    ),
]

Performance Optimization

Parallel Node Execution

# Use multiple edges from same source for parallel execution
edges = [
    EdgeSpec(
        id="start-to-search1",
        source="start",
        target="search-source-1",
        condition=EdgeCondition.ALWAYS,
    ),
    EdgeSpec(
        id="start-to-search2",
        source="start",
        target="search-source-2",
        condition=EdgeCondition.ALWAYS,
    ),
    EdgeSpec(
        id="start-to-search3",
        source="start",
        target="search-source-3",
        condition=EdgeCondition.ALWAYS,
    ),
    # Converge results
    EdgeSpec(
        id="search1-to-merge",
        source="search-source-1",
        target="merge-results",
    ),
    EdgeSpec(
        id="search2-to-merge",
        source="search-source-2",
        target="merge-results",
    ),
    EdgeSpec(
        id="search3-to-merge",
        source="search-source-3",
        target="merge-results",
    ),
]

Handoff to Testing

When agent is complete, transition to testing phase:

print("""
✅ Agent complete: exports/my_agent/

Next steps:
1. Switch to testing-agent skill
2. Generate and approve tests
3. Run evaluation
4. Debug any failures

Command: "Test the agent at exports/my_agent/"
""")

Pre-Testing Checklist

Before handing off to testing-agent:

• Agent structure validates: python -m agent_name validate
• All nodes defined in nodes/init.py
• All edges connect valid nodes
• Entry node specified
• Agent can be imported: from exports.agent_name import default_agent
• README.md with usage instructions
• CLI commands work (info, validate)

Related Skills

• building-agents-core - Fundamental concepts
• building-agents-construction - Step-by-step building
• testing-agent - Test and validate agents
• agent-workflow - Complete workflow orchestrator

Remember: Agent is actively constructed, visible the whole time. No hidden state. No surprise exports. Just transparent, incremental file building.