Workspace/Mini projects
Loading progress
All mini projects
MODULE 32 · 16 HOUR BUILD

Evidence desk: an integrated agent capstone

Deliver a tenant-aware incident evidence desk with a usable task interface, an MCP capability service, an A2A specialist integration, evaluation, operational controls, and a truthful repository report.

Build evidence Record your actual checks, results, and limitations.

Build it in stages

  1. Run the local seed and write the user-task contract, permissions, failure states, and baseline.
  2. Replace the retrieval boundary with an actual MCP adapter using a pinned protocol and SDK, then verify scoped resources and tool results.
  3. Implement an A2A specialist service and client with capability discovery, bounded tasks, artifact validation, cancellation, and documented version compatibility.
  4. Build a task interface showing progress, evidence, missing information, and concrete action review where required.
  5. Add paired task evaluation, synthetic adversarial regressions, trace correlation, budget enforcement, and a tested rollback or disable path.
  6. Publish a repository package with a quick start, synthetic fixture, exact measured results, limitations, architecture decisions, and resume bullets supported by those artifacts.

Your acceptance criteria

Use these as your project review. Record commands, outputs, and failure cases in your repository.

  • The seed remains runnable offline; live integrations are separately labeled and tested against their pinned protocols.
  • The finished workflow completes a cited incident-review task through both actual protocol adapters and the interface.
  • Tenant isolation, duplicate events, unavailable peers, exhausted budgets, and cancellation have reproducible checks.
  • Every final citation references authorized evidence, with a separate semantic-support evaluation.
  • Baseline and candidate results include sample size, task definitions, resource budget, and failures.
  • README claims distinguish simulations, actual protocol tests, measured model results, and unmeasured production behavior.

A working starting point

The seed runs as supplied. Extend it to satisfy the full brief. It is a teaching starting point, not a finished portfolio submission.

main.py
python
import html
import json

RECORDS = [
    {"id": "red-1", "tenant": "red", "text": "Latency increased with queue depth."},
    {"id": "red-2", "tenant": "red", "text": "Worker CPU remained stable."},
    {"id": "blue-1", "tenant": "blue", "text": "CANARY_BLUE_PRIVATE"},
]

def retrieval_adapter(tenant):
    return [dict(row) for row in RECORDS if row["tenant"] == tenant]

def specialist_adapter(evidence):
    if not evidence:
        return {"summary": "No authorized evidence is available.", "citations": []}
    return {"summary": "Review queue growth and worker capacity before changing the service.",
            "citations": [row["id"] for row in evidence]}

def render(artifact, state):
    citations = "".join(f"<li>{html.escape(value)}</li>" for value in artifact["citations"])
    return (f"<article><h1>Incident review</h1><p>{html.escape(state)}</p>"
            f"<p>{html.escape(artifact['summary'])}</p><ul>{citations}</ul></article>")

def run_task(tenant, budget):
    events = [{"seq": 1, "state": "queued", "cost": 0}]
    if budget < 3:
        events.append({"seq": 2, "state": "failed", "cost": 0})
        return {"state": "failed", "reason": "insufficient budget", "events": events}
    events.append({"seq": 2, "state": "working", "cost": 1})
    evidence = retrieval_adapter(tenant)
    artifact = specialist_adapter(evidence)
    permitted = {row["id"] for row in evidence}
    valid = bool(artifact["citations"]) and set(artifact["citations"]) <= permitted
    state = "completed" if valid else "failed"
    events.append({"seq": 3, "state": state, "cost": 2})
    return {"state": state, "artifact": artifact, "events": events,
            "cost_units": sum(event["cost"] for event in events),
            "ui_html": render(artifact, state), "citation_membership_valid": valid}

if __name__ == "__main__":
    result = run_task("red", 3)
    assert result["state"] == "completed"
    assert result["cost_units"] == 3
    assert "blue-1" not in result["artifact"]["citations"]
    assert run_task("red", 2)["state"] == "failed"
    assert run_task("unknown", 3)["state"] == "failed"
    print(json.dumps({"scope": "local contract simulation; no MCP/A2A transport or model",
                      "result": result}, indent=2, sort_keys=True))

Push it further

Conduct a small authorized user study with predefined task outcomes, compare the integrated workflow with its simpler baseline, and publish conditional findings including cases where delegation is not worth its cost.