Agentic Engineering Lab

Build it in stages

Run the seed and document the request and result JSON contracts.
Add a strict reusable record validator and invalid fixtures for every field.
Move fixture reads behind an async transport interface with a configurable concurrency limit.
Add timeout and expected-error categories without hiding programmer exceptions.
Create a manifest with code revision, dependency identity, input fingerprint, and non-secret configuration.
Expose the envelope through a small HTTP adapter and document a frontend fetch example.

Your acceptance criteria

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

A four-item fixture produces three accepted records and one explicit failure.
Duplicate normalized IDs, booleans, empty IDs, and unknown fields have deterministic outcomes.
A trace demonstrates active requests never exceed the configured limit.
A fresh checkout can run the offline seed without credentials.
The response never contains provider credentials or raw exception tracebacks.

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 asyncio
import hashlib
import json

BODIES = [
    '{"id":"A","quantity":2}',
    '{"id":"B","quantity":4}',
    '{"id":"C","quantity":true}',
    '{"id":"D","quantity":1}',
]

def validate(body):
    row = json.loads(body)
    if not isinstance(row, dict) or set(row) != {"id", "quantity"}:
        raise ValueError("invalid_shape")
    if not isinstance(row["id"], str) or not row["id"].strip():
        raise ValueError("invalid_id")
    if type(row["quantity"]) is not int or row["quantity"] < 1:
        raise ValueError("invalid_quantity")
    return {"id": row["id"].strip(), "quantity": row["quantity"]}

async def run():
    gate = asyncio.Semaphore(2)
    async def ingest(index, body):
        async with gate:
            await asyncio.sleep(0)
            try:
                return {"index": index, "status": "accepted", "record": validate(body)}
            except (ValueError, TypeError) as error:
                return {"index": index, "status": "rejected", "error": str(error)}
    results = await asyncio.gather(*(ingest(i, body) for i, body in enumerate(BODIES)))
    raw = json.dumps(BODIES, separators=(",", ":")).encode()
    return {"schema_version": 1, "status": "completed",
            "accepted": sum(r["status"] == "accepted" for r in results),
            "input_sha256": hashlib.sha256(raw).hexdigest(), "results": results}

if __name__ == "__main__":
    print(json.dumps(asyncio.run(run()), indent=2))

Push it further

Add a bounded producer-consumer queue and compare memory behavior against creating one task per item for a large generated input.