Skip to content

Strands-Coder: Architecture of an Autonomous Agent 🏗️

Published: January 3, 2025

Strands-Coder is an autonomous AI agent that runs 24/7 on GitHub, contributing to the strands-agents ecosystem. This post dives deep into its architectural decisions.

High-Level Architecture

flowchart TB
    subgraph "Triggers"
        GHE[GitHub Events]
        CRON[Scheduled Cron]
        WD[Workflow Dispatch]
    end

    subgraph "Entry Point"
        ACTION[action.yml]
        RUNNER[agent_runner.py]
    end

    subgraph "Context Layer"
        CTX[context.py]
        SELF[Self-Awareness]
        PROJECT[Project Context]
        EVENT[Event Context]
    end

    subgraph "Agent Core"
        STRANDS[Strands Agent]
        MODEL[Model Provider]
        TOOLS[Tool Registry]
    end

    subgraph "Persistence"
        KB[(Knowledge Base)]
        S3[(S3 Sessions)]
        VARS[(GitHub Variables)]
    end

    GHE --> ACTION
    CRON --> ACTION
    WD --> ACTION
    ACTION --> RUNNER
    RUNNER --> CTX
    CTX --> SELF
    CTX --> PROJECT
    CTX --> EVENT
    CTX --> STRANDS
    STRANDS <--> MODEL
    STRANDS <--> TOOLS
    STRANDS <--> KB
    STRANDS <--> S3
    STRANDS <--> VARS

    style STRANDS fill:#f9f,stroke:#333,stroke-width:4px

Design Decision #1: Composite GitHub Action

# action.yml - The entry point
runs:
  using: 'composite'
  steps:
    - name: Checkout
    - name: Configure AWS (Bedrock only)
    - name: Set up Python
    - name: Install dependencies
    - name: Download Custom Agent Runner  # NEW!
    - name: Agent

Why Composite?

  • Reusable: Any repo can use uses: cagataycali/strands-coder@main
  • Self-contained: All dependencies in one place
  • Extensible: Custom agent runners via URL
  • Cacheable: uv cache for fast installs

Custom Agent Runner Pattern

inputs:
  agent_runner:
    description: 'Custom agent runner URL'
    required: false

This allows anyone to inject their own agent logic:

- uses: cagataycali/strands-coder@main
  with:
    prompt: "Custom task"
    agent_runner: "https://gist.githubusercontent.com/.../custom_agent.py"

Design Decision #2: Rich Context Building

sequenceDiagram
    participant GH as GITHUB_CONTEXT
    participant API as GitHub GraphQL
    participant CTX as context.py
    participant Agent

    GH->>CTX: Raw event JSON
    CTX->>CTX: Detect event type

    alt Issue Event
        CTX->>API: Fetch issue + comments
        CTX->>API: Fetch linked PRs
    else PR Event
        CTX->>API: Fetch reviews
        CTX->>API: Fetch review threads
        CTX->>API: Fetch linked issues
    else Discussion Event
        CTX->>API: Fetch replies
    end

    CTX->>CTX: Fetch project state
    CTX->>CTX: Inject self code
    CTX->>Agent: Complete system prompt

The Context Builder

def build_system_prompt() -> str:
    """Build comprehensive system prompt from environment and context."""
    base_prompt = os.getenv("SYSTEM_PROMPT", "")

    # Add rich GitHub event context
    github_event_context = fetch_github_event_context()

    # Add Project context
    project_context = fetch_project_context(project_id)

    # Add self-awareness
    own_code = get_own_source_code()

    return f"{base_prompt}\n\n{github_event_context}\n\n{project_context}\n\n{own_code}"

Why This Matters

The agent doesn't just see "issue #42 created" - it sees:

  • The full issue body
  • All previous comments
  • Linked PRs
  • Project board status
  • Its own implementation

Design Decision #3: User Message Extraction

def extract_user_message() -> str:
    """Extract the actual user message for semantic matching."""
    event_name = github_context.get("event_name")

    if event_name == "issue_comment" and action == "created":
        return event.get("comment", {}).get("body", "")
    elif event_name == "issues" and action == "opened":
        return event.get("issue", {}).get("body", "")
    # ... more event types

This extracted message is used for: 1. Knowledge base retrieval: retrieve(text=user_message, ...) 2. Enhanced prompts: Appended to the main prompt for context

Design Decision #4: Flexible Tool Loading

def load_tools(config: str) -> list:
    """
    Format: package1:tool1,tool2;package2:tool3,tool4
    """
    tools = []
    for group in config.split(";"):
        package, tools_str = group.split(":", 1)
        for tool_name in tools_str.split(","):
            module = __import__(package, fromlist=[tool_name])
            tool = getattr(module, tool_name)
            tools.append(tool)
    return tools

graph LR
    Config[STRANDS_TOOLS] --> Parse
    Parse --> P1[strands_tools:shell,retrieve]
    Parse --> P2[strands_coder:use_github,projects]
    P1 --> Load1[Import shell]
    P1 --> Load2[Import retrieve]
    P2 --> Load3[Import use_github]
    P2 --> Load4[Import projects]
    Load1 & Load2 & Load3 & Load4 --> Registry[Tool Registry]

Tool Configuration

# Default tools
STRANDS_TOOLS="strands_tools:shell,retrieve;strands_coder:use_github,projects"

# Minimal for fast runs
STRANDS_TOOLS="strands_tools:shell"

# Full power
STRANDS_TOOLS="strands_tools:shell,retrieve,slack;strands_coder:use_github,create_subagent,system_prompt,store_in_kb,projects,scheduler"

Design Decision #5: MCP Server Integration

def load_mcp_servers() -> list:
    """Load MCP servers from MCP_SERVERS env var."""
    config = json.loads(os.getenv("MCP_SERVERS", "{}"))

    for name, cfg in config.get("mcpServers", {}).items():
        if cfg.get("disabled"):
            continue

        # Handle disabled tools per server
        disabled_tools = cfg.get("disabledTools", [])
        tool_filters = ToolFilters(rejected=disabled_tools)

        if "command" in cfg:
            # stdio transport
            client = MCPClient(transport_callable=..., tool_filters=tool_filters)
        elif "url" in cfg:
            # HTTP/SSE transport
            client = MCPClient(transport_callable=..., tool_filters=tool_filters)

This enables composable agent capabilities:

{
  "mcpServers": {
    "strands-docs": {
      "command": "uvx",
      "args": ["strands-agents-mcp-server"],
      "disabledTools": ["fetch_doc"]  // Only use search
    }
  }
}

Design Decision #6: Trace Linking

def extract_issue_id() -> str:
    """Extract issue ID for trace linking."""
    if event_name == "issue_comment":
        return str(event.get("issue", {}).get("number"))
    elif event_name == "pull_request":
        return f"pr-{event.get('pull_request', {}).get('number')}"

Every trace is linked to its triggering issue/PR:

agent = Agent(
    trace_attributes={
        "session.id": session_id,
        "issue_id": issue_id,  # CRUCIAL
        "tags": ["Strands-Agents", "GitHub-Action", f"issue:{issue_id}"]
    }
)

graph LR
    Issue[Issue #42] --> Trace1[Trace A]
    Issue --> Trace2[Trace B]
    Issue --> Trace3[Trace C]

    Trace1 --> LF[Langfuse Dashboard]
    Trace2 --> LF
    Trace3 --> LF

    LF --> Filter[Filter by issue:42]
    Filter --> Analysis[Debug Agent Behavior]

Design Decision #7: Scheduler Tool

stateDiagram-v2
    direction LR

    [*] --> AGENT_SCHEDULES

    AGENT_SCHEDULES --> ControlLoop: Hourly Check
    ControlLoop --> ParseJobs
    ParseJobs --> CheckCron
    CheckCron --> MatchNow: Cron Matches
    CheckCron --> Skip: No Match
    MatchNow --> Dispatch
    Dispatch --> UpdateTimestamp
    UpdateTimestamp --> AGENT_SCHEDULES

    note right of AGENT_SCHEDULES
        GitHub Variable storing
        all scheduled jobs as JSON
    end note
# scheduler.py - Cron-based job scheduling via GitHub Variables
@tool
def scheduler(
    action: str,
    job_id: str = None,
    cron: str = None,      # "0 9 * * *" = daily at 9 AM
    run_at: str = None,    # "2024-01-20T14:00:00Z" = one-time
    once: bool = False,    # Auto-remove after execution
    prompt: str = None,
    system_prompt: str = None,
    tools: str = None,
) -> dict:

Why a custom scheduler?

  • GitHub's cron is limited (5 schedules max)
  • Agent can self-schedule based on discoveries
  • Supports one-time jobs with auto-cleanup
  • Configuration stored in version control

Design Decision #8: Project Integration

@tool  
def projects(
    action: str,
    project_id: str = None,  # Defaults to STRANDS_CODER_PROJECT_ID
    item_id: str = None,
    field_name: str = None,
    field_value: str = None,
) -> dict:

The agent tracks ALL its work on a GitHub Project board:

kanban
    Todo
        Issue #1
        Issue #5
    In Progress
        PR #42
    Done
        Issue #3

Every execution: 1. Check status: projects(action="get_progress") 2. Update items: Move work to "In Progress" 3. Clean up: Remove "Done" items

Design Decision #9: Exit Strategy

# agent_runner.py
if has_mcp_servers and not os.getenv("PYTEST_CURRENT_TEST"):
    os._exit(0)  # Force kill background threads
else:
    sys.exit(0)  # Normal exit

MCP Background Threads

MCP servers run in daemon threads. sys.exit() won't kill them. os._exit() ensures clean GitHub Actions completion.

The Complete Flow

sequenceDiagram
    participant GH as GitHub
    participant GA as GitHub Actions
    participant AR as agent_runner.py
    participant CTX as context.py
    participant Agent as Strands Agent
    participant KB as Knowledge Base

    GH->>GA: Event (issue/PR/cron)
    GA->>AR: Run composite action
    AR->>AR: Setup OTEL
    AR->>AR: Load tools
    AR->>AR: Load MCP servers
    AR->>CTX: Build system prompt
    CTX->>CTX: Fetch event context
    CTX->>CTX: Fetch project context
    CTX->>CTX: Inject self code
    CTX->>AR: Complete prompt
    AR->>Agent: Create with prompt

    Agent->>KB: retrieve(user_message)
    KB->>Agent: Related context
    Agent->>Agent: Execute task
    Agent->>KB: store_in_kb(summary)

    AR->>GA: Write summary
    AR->>AR: Exit (os._exit if MCP)

Architecture is about making the right trade-offs. Strands-Coder prioritizes autonomy, observability, and continuous learning. 🏗️