Chapter 9 — Building an Agent That Uses the Rest of the Code Craft AI Portfolio

Eight chapters established the SDK. This chapter watches the SDK in motion — an agent that runs sysman features against a landscape through interop protocols, with full event recording for time-travel debugging. Every subsystem the SDK exposes lands in this case study.

The volume's payoff. A worked example of an agent that uses the framework's other tiers — sysman from Volume 4, landscape from Volume 6, interop protocols from Volume 5, testbase from Volume 1. By the end, a reader should be able to take the case study's shape and apply it to their own automation domain.

The case study is realistic. An agent that triages production incidents — when an alert fires, the agent gathers context from production hosts, queries logs, runs diagnostics, summarises findings, and either fixes the issue or escalates to a human via approval gate.

The scenario

A small operations team receives ~10 alerts per day across their fleet of ~200 production hosts. Each alert needs:

1. Context gathered (logs, metrics, recent deployments).
2. Diagnostics run (system state checks).
3. Severity assessed.
4. Either a fix applied or escalated to humans.

The team's SLA is 15 minutes for triage. Manually, this takes 10–15 minutes per alert. An AI-driven triage agent could handle low-severity alerts autonomously and produce rich context for high-severity ones.

The agent is not trying to replace human judgement. It's trying to make the human's job easier — present what matters, suppress what doesn't, propose actions.

The agent's shape

from ccai.agentkit.agent import Agent
from ccai.agentkit.agentconfiguration import AgentConfiguration
# ... other imports ...


class IncidentTriageAgent(Agent):
    """
        Triages production incidents.  Gathers context, runs
        diagnostics, summarises, escalates when severity exceeds
        an autonomous-action threshold.
    """

    def initialize(self) -> None:
        super().initialize()

        # LLM provider with circuit breaker
        self._configure_llm_providers()

        # Tools — sysman feature wrappers, log queries, etc.
        self._configure_tools()

        # Memory store for retaining incident-history context
        self._configure_memory()

        # Workflow definition
        self._workflow = self._build_workflow()

        # Approval gate for autonomous-action threshold
        self._approval_gate = ApprovalGate()

        # Observability — recorder for time-travel debugging
        self._execution_recorder = ExecutionRecorder(
            agent_id=self.identity.identifier,
        )
        self._execution_recorder.register_sink(
            FileExecutionSink(path="recordings/")
        )

        return

    def start(self) -> None:
        super().start()
        # Wait for an incident; when one arrives, run the workflow.
        self._wait_for_incident()

Eight subsystems wired in initialize():

• LLM providers (chapter 2).
• Tool registry (chapter 3).
• Memory store (chapter 4).
• Workflow definition (chapter 5).
• Approval gate (chapter 5).
• Circuit breaker (chapter 5).
• Execution recorder (chapter 6).
• Cost tracker (built into the Agent base, chapter 1).

The tool surface — sysman feature wrappers

The agent needs to do things on production hosts:

• Read system state (free disk, running services, recent log entries).
• Run sysman feature queries (Volume 4's read verbs).
• Apply sysman feature changes (Volume 4's write verbs) when authorised.

Each capability is a tool (chapter 3). A representative set:

class GetServiceStatusTool(Tool):
    """
        Query the status of a service on a production host.
    """
    TOOL_NAME = "get_service_status"
    TOOL_DESCRIPTION = "Get the status of a service on a host."
    TOOL_VERSION = "1.0.0"
    INPUT_SCHEMA = {
        "parameters": [
            {"name": "host_name", "type": "string", "required": True},
            {"name": "service_name", "type": "string", "required": True},
        ]
    }
    OUTPUT_SCHEMA = {
        "parameters": [
            {"name": "host_name", "type": "string", "required": True},
            {"name": "service_name", "type": "string", "required": True},
            {"name": "active_state", "type": "string", "required": True},
            {"name": "load_state", "type": "string", "required": True},
        ]
    }

    def __init__(self, landscape: "Landscape"):
        super().__init__()
        self._landscape = landscape
        return

    def execute(self, inputs: dict, credentials: dict = None) -> dict:
        # Look up the host's device from the landscape (Vol 6 ch.3).
        device = self._landscape.device(inputs["host_name"])

        # The landscape's per-OS sysman ServiceManager (Vol 4 ch.5,
        # ch.6, ch.7) — the right ServiceManager for the host's OS.
        service_mgr = device.sysman.service_manager

        # Sysman read verb (Vol 4 ch.3 — reads never elevate).
        info = service_mgr.query(inputs["service_name"])

        return {
            "host_name": inputs["host_name"],
            "service_name": inputs["service_name"],
            "active_state": info.active_state,
            "load_state": info.load_state,
        }

The tool is thin. It converts agent-level inputs to sysman-level calls and back. The actual work — querying the service via SSH, parsing systemctl output — happens in the sysman tier (Volume 4) over the interop wire (Volume 5).

A typical agent ships dozens of such tool wrappers, covering the sysman surface the agent needs. The wrappers are application-specific; the SDK provides the patterns; the application chooses which capabilities to expose.

Tools that don't elevate vs tools that do

A subtle point worth pinning down. Volume 4's contract discipline says reads never elevate; writes elevate. The agent's tool design respects this:

• Read tools (GetServiceStatusTool) construct a ServiceManager with config.elevated=False (the default).
• Write tools (RestartServiceTool) construct a ServiceManager with config.elevated=True only when the agent has been granted authority.

Authority comes from the agent's configuration — typically a per-run flag like autonomous_actions=False (default) that the operator overrides for specific high-confidence runs.

The workflow

The agent's workflow handles an incident in five steps:

def _build_workflow(self) -> WorkflowDefinition:
    return WorkflowDefinition(
        name="incident_triage",
        steps=[
            WorkflowStep(step_id="gather_context",
                         step_callable=self._step_gather_context),
            WorkflowStep(step_id="run_diagnostics",
                         step_callable=self._step_run_diagnostics),
            WorkflowStep(step_id="assess_severity",
                         step_callable=self._step_assess_severity),
            WorkflowStep(step_id="propose_action",
                         step_callable=self._step_propose_action),
            WorkflowStep(step_id="execute_or_escalate",
                         step_callable=self._step_execute_or_escalate),
        ],
        edges=[
            WorkflowEdge(from_step="gather_context", to_step="run_diagnostics"),
            WorkflowEdge(from_step="run_diagnostics", to_step="assess_severity"),
            WorkflowEdge(from_step="assess_severity", to_step="propose_action"),
            WorkflowEdge(from_step="propose_action", to_step="execute_or_escalate"),
        ],
    )

Linear DAG. Five steps, one path through. Each step:

1. gather_context — query Volume 6's landscape for the host's group, role, recent events. Pull recent log entries via Volume 5's SSH wire.
2. run_diagnostics — run targeted diagnostic commands via Volume 4's sysman feature managers.
3. assess_severity — call the LLM with the gathered context; LLM returns a severity score 1–5.
4. propose_action — call the LLM with the severity and context; LLM proposes specific actions.
5. execute_or_escalate — branch on severity and the agent's authorisation:
   • Severity ≤ 2 with autonomous_actions=True → execute via sysman write verbs.
   • Severity ≥ 3 OR autonomous_actions=False → escalate via approval gate.

The workflow is deterministic in shape — same five steps every run. The variation lives in the LLM responses (severity score, proposed action) and the conditional in step 5.

A worked step — gather_context

def _step_gather_context(self, context: WorkflowContext) -> dict:
    """
        Step 1 — Gather context for the incident.
    """
    incident = context.get("incident")
    host_name = incident["host_name"]

    # Pull recent log entries.
    log_tool = self._tool_registry.lookup("read_recent_logs")
    log_result = self._tool_executor.execute(
        log_tool,
        inputs={"host_name": host_name, "lines": 200},
    )

    # Pull host metadata from the landscape.
    device = self._landscape.device(host_name)

    # Pull recent deployments from the team's deployment service.
    deployment_tool = self._tool_registry.lookup("recent_deployments")
    deploy_result = self._tool_executor.execute(
        deployment_tool,
        inputs={"host_name": host_name, "since": incident["alert_timestamp"]},
    )

    context.set("logs", log_result.output["lines"])
    context.set("device_info", {
        "host": device.host,
        "os": device.os,
        "role": device.role,
    })
    context.set("recent_deployments", deploy_result.output["deployments"])

    return {"context_gathered": True}

Three tool calls, three pieces of context, all stored in the workflow's context for downstream steps. The method on the agent class wires the tools through the executor; the workflow sequences the methods.

A worked step — execute_or_escalate

def _step_execute_or_escalate(self, context: WorkflowContext) -> dict:
    """
        Step 5 — Either execute the proposed action or escalate.
    """
    severity = context.get("severity")
    proposed_action = context.get("proposed_action")
    autonomous_threshold = self._configuration.autonomous_threshold

    if severity <= autonomous_threshold:
        # Execute autonomously.
        return self._execute_action(proposed_action, context)
    else:
        # Escalate via approval gate.
        return self._escalate_for_approval(proposed_action, context)


def _execute_action(self, action: dict, context: WorkflowContext) -> dict:
    """
        Run the proposed action through sysman feature managers.
    """
    device = self._landscape.device(action["host_name"])
    if action["type"] == "restart_service":
        # Sysman write verb (Vol 4 ch.5/ch.6/ch.7 ServiceManager).
        result = device.sysman.service_manager.restart(action["service_name"])
        return {"action_executed": True, "sysman_result": result}
    # ... other action types ...


def _escalate_for_approval(self, action: dict, context: WorkflowContext) -> dict:
    """
        Create an approval request and wait for human resolution.
    """
    request = self._approval_gate.create_request(
        execution_id=context.execution_id,
        step_id="execute_or_escalate",
        description="Approve action: {} on {}".format(
            action["type"], action["host_name"]
        ),
        requesting_agent_id=self.identity.identifier,
        metadata={"proposed_action": action, "context": context.as_dict()},
    )

    # Notify on-call (out-of-band — Slack, email, etc.).
    self._notify_oncall(request)

    # Block on resolution.
    resolution = request.wait_for_resolution(timeout=900)  # 15 min

    if resolution == "approved":
        return self._execute_action(action, context)
    elif resolution == "rejected":
        return {"action_skipped": True, "reason": request.rejection_reason}
    elif resolution == "expired":
        return {"action_skipped": True, "reason": "approval_timeout"}

The branch:

• Severity ≤ threshold → autonomous action via sysman write verbs.
• Severity > threshold → escalate through the approval gate to an on-call human.

The cost tracker in action

Throughout the run, every LLM call records its cost. At the end:

def _on_workflow_complete(self, result: WorkflowResult) -> None:
    cost = self._cost_tracker.total_cost_estimate
    self.logger.info(
        "Triage complete. Cost: ${:.4f} ({} prompt + {} completion tokens)".format(
            cost.total_cost,
            self._cost_tracker.total_token_usage.prompt_tokens,
            self._cost_tracker.total_token_usage.completion_tokens,
        )
    )

A typical incident triage might use:

• ~3,000 prompt tokens (system prompt + gathered context).
• ~500 completion tokens (severity assessment).
• ~3,500 prompt tokens (system prompt + action proposal).
• ~1,000 completion tokens (proposed action).

Total: ~8,000 tokens per incident, ~$0.05 at typical GPT-4-class pricing. At 10 incidents per day, $0.50 per day, $15 per month per agent.

The cost is visible. The team's monthly review pulls totals from the cost tracker (perhaps aggregated to a dashboard via the metrics layer from chapter 6).

Observability — recording for replay

Every event in the run is captured by the ExecutionRecorder:

• Lifecycle transitions.
• Workflow step boundaries.
• Tool calls (started / completed / failed).
• LLM calls (started / completed / failed).
• Approval gate events.
• The agent's own custom events (e.g., "alerts received," "incident closed").

After the run, the recording is persistent. A debugger loads the recording via ExecutionReplayer and walks through it step by step:

recording = ExecutionRecording.load("recordings/triage-20260501-1432.json")
replayer = ExecutionReplayer(recording=recording)

while replayer.has_next:
    record = replayer.next()
    print(f"[{record.sequence_number}] {record.event_type}")
    if record.event_type == "tool.execution.completed":
        print(f"  Tool: {record.payload['tool_name']}")
        print(f"  Duration: {record.payload['duration_ms']}ms")
        print(f"  Output: {record.payload['output']}")
    elif record.event_type == "llm.request.completed":
        print(f"  Provider: {record.payload['provider_name']}")
        print(f"  Tokens: {record.payload['token_usage']}")

The replay is the bridge to Volume 8 (time-travel debugging). Volume 8 walks the production-grade post-mortem tooling that consumes recordings to reconstruct what happened during a past run.

Integration with the rest of the framework

Stepping back, this case study touches every drafted volume:

• Volume 0 — error vocabulary, configuration, application identity (the agent has all of these).
• Volume 1 — testbase tests for every tool wrapper, workflow step, and integration point (chapter 7's mock- first discipline applies).
• Volume 2 — when the agent runs as a distributed workload, the tandem-validator pattern from Vol 2 ch.5 applies for monitoring its behaviour.
• Volume 3 — taskplus orchestration that invokes the agent (Vol 3 ch.25's bridge tasks) for scheduled triage runs.
• Volume 4 — every sysman tool wrapper goes through Volume 4's contract layer.
• Volume 5 — every wire call (SSH, REST) goes through Volume 5's Measured Interoperability discipline.
• Volume 6 — the landscape provides the device metadata; the topology provides the cluster membership the agent reasons about.
• Volume 7 (this volume) — the SDK ties them all together.

The case study is the volume's proof that the SDK's patterns compose with the rest of the portfolio. An agent that's built on the SDK gets:

• Predictable tool dispatch (chapter 3's validate-resolve-execute-validate).
• Predictable LLM routing with failover (chapter 2).
• Predictable memory and conversation handling (chapter 4).
• Predictable workflow execution (chapter 5).
• Predictable observability (chapter 6).
• Predictable testing through testbase mock patterns (chapter 7).
• Predictable property-test coverage (chapter 8).

Eight predictabilities. An agent built outside the SDK has to invent each. Building on the SDK means inheriting each.

What this volume accomplished

Looking back across the nine chapters:

• Chapter 1 — SDK vs framework, four production-agent concerns, the agent state machine.
• Chapter 2 — multi-provider LLM abstraction with failover.
• Chapter 3 — tool design with custom schema validation.
• Chapter 4 — memory architecture with three scopes and semantic search.
• Chapter 5 — DAG-based workflow with approval gates and circuit breakers.
• Chapter 6 — three observability concerns: tracing, metrics, replay.
• Chapter 7 — 305-test mock-first testing discipline.
• Chapter 8 — 72-property-test invariant coverage.
• Chapter 9 — the case study tying it all together.

A reader who's read the volume can:

• Build a production agent on the SDK.
• Choose which subsystems they need; the SDK is composable.
• Test the agent with the same testbase machinery the rest of the framework uses.
• Observe the agent's behaviour through the standard stack.
• Integrate the agent with the rest of the codecraftai-* portfolio.

End of Volume 7

The series's largest volume by source-material size and test-suite scale. The agent-kit is ~46k LOC, ~217 modules, 305 test files. The volume's nine chapters covered the SDK's structure end-to-end, with a worked case study as the payoff.

The chapters that come next are in the volume that follows. Volume 8 walks time-travel debugging in production — the production-grade post-mortem tooling that consumes the ExecutionRecording artifacts this volume's chapter 6 introduced. Where Volume 7 said "every run produces a recording," Volume 8 says "here's what to do with the recordings when something goes wrong."

Volume 7's contribution to the series is the AI-agent discipline. The SDK's structure — base classes, custom schemas, structured exceptions, mock-first tests, property tests at scale — is the framework's answer to "how do you build production AI agents that don't surprise you?" Every other volume's contribution to the agent's correctness is mediated through this volume's patterns.