An agent result can have the right JSON shape and still misread a source. This guide gives a Python developer a complete route from one permitted notes file to a local action register: a typed producer result, source and meaning checks, exact-hash human approval, and a consumer that writes and reads back a grouped Markdown report. Use your own notes containing explicit commitments. No real notes or model result were used to prepare this guide.

The big change

  • What changed: The OpenAI Agents SDK accepts a Pydantic output_type, so an application can ask an agent for a defined result and check it again before using it. This example adds source citations and a versioned contract around that result.
  • Why it matters: A developer connecting an agent to another work step can stop malformed, incomplete or unsupported output before it becomes an action list. A reviewer still has to judge whether the quoted notes actually support every task, owner and date, and whether the model omitted anything.
  • What to watch: The decision point is the consumer boundary. This recipe produces a local triage register only after approval of the exact run, source and candidate. A later task-system import would need its own authority, retry policy and destination readback.

Prepare one authorized source

You need Python 3.12, an OpenAI API account authorized for the notes, a reviewer who can read the original file, and a model you have confirmed currently supports JSON-schema structured output in your account. Check the account's data terms and current pricing before running the producer. Do not put credentials in the notes or approval file. The only model call below is produce; the rest is local processing.

Save your original notes as UTF-8 at notes.txt. They must contain actual explicit action commitments from your own work. Assign a stable source_id, such as your own record identifier, and keep the file byte-for-byte unchanged until review and consumption. Create candidate/ before producing; the candidate file must be new. Make a separate output directory available for the later local report.

Record the source's byte hash before the run:

Terminal
python -c "from pathlib import Path; import hashlib; print(hashlib.sha256(Path('notes.txt').read_bytes()).hexdigest())"

In an isolated Python 3.12 environment, create recipe-requirements.txt with these exact lines and install it:

Text
openai-agents==0.24.0
pydantic==2.14.1
Terminal
python -m pip install -r recipe-requirements.txt
mkdir candidate

Those versions installed together in a local Python 3.12.7 environment on October 20, 2026. Confirm compatibility and model availability when you run the guide. Save the complete script below as action_register_recipe.py in the same directory. It has two commands: produce makes a candidate and run receipt; consume builds the report only after validation and approval.

Run the typed producer

The Extraction contract requires action-register/v1, ready, partial or blocked status, the source ID and SHA-256 of the exact file bytes, actions and a gap_note. Each action requires text, an exact quotation and 1-based source line span, nullable owner and date, and complete or needs_review evidence status. A null owner or date means the notes did not state it. The model is instructed not to infer either. All fields are required and extra fields are rejected.

Choose the structured-output model explicitly. Supply your own permitted path and identifier:

Terminal
python action_register_recipe.py produce --source-id OWNED-ID --source notes.txt --candidate candidate/actions.json --model YOUR-STRUCTURED-OUTPUT-MODEL

Read candidate/actions.json beside candidate/actions.run.json. The latter records the UUID run ID, source and candidate hashes, finishing agent and requested model. A failure instead creates a run-ID-suffixed .failure.json diagnostic record. An interruption, None or plain-string final output, or an unexpected finishing agent, never counts as success. This command does not persist the SDK's resumable RunState; its failure receipt cannot resume the run. Inspect the request, candidate and any downstream state before deciding whether a new run is appropriate.

Review meaning and approve the exact candidate

The application checks the candidate's strict types and required fields, source identity, ready status, at least one complete action, valid line spans, quotation presence in the cited span, nonblank task or owner, and canonical YYYY-MM-DD dates. That proves structure and quotation placement. It cannot establish that a sentence is truly a commitment, that a name or date means what the candidate says, or that all commitments were found.

Read the original notes and every proposed action. Compare task meaning, owner, date, quote and omissions. Resolve ambiguity in the source or candidate through a new reviewed run; leave unsupported, partial or blocked candidates unapproved. After checking that the unchanged file and candidate match actions.run.json, write approval.json with exactly these seven keys. Copy the run ID and hashes from the checked receipt. Use the reviewer's real identifier and a substantive account of what was checked:

JSON
{
  "approved": true,
  "run_id": "COPY-RUN-ID",
  "source_id": "OWNED-ID",
  "source_sha256": "COPY-SOURCE-SHA256",
  "payload_sha256": "COPY-PAYLOAD-SHA256",
  "reviewer": "YOUR-REVIEWER-ID",
  "review_note": "Describe the source, task, owner, date, quotation and omission checks actually performed."
}

approved must be the JSON boolean true, not a string or number. All other values must be strings; reviewer and note cannot be blank. Additional fields fail. The approval binds one candidate and source revision, not any future rewrite.

Consume and read back the report

Terminal
python action_register_recipe.py consume --source-id OWNED-ID --source notes.txt --candidate candidate/actions.json --approval approval.json --out private-output

The consumer groups accepted actions into Assigned with dates, Needs an owner and Needs a date. Each entry contains its task, owner/date state, quotation and source lines. It writes action-register-<payload-hash-prefix>.md and a neighboring .receipt.json with run, source, candidate and report hashes. Open both files, compare the report's cited actions with the notes, and verify report_sha256 against the Markdown file's bytes. A valid receipt says only that the local artifact was created and read back; it does not approve an external task-system write.

The consume command prints the actual report and receipt paths. Use those paths for this byte check:

Terminal
python -c "from pathlib import Path; import hashlib,json,sys; report=Path(sys.argv[1]); receipt=json.loads(Path(sys.argv[2]).read_text(encoding='utf-8')); assert hashlib.sha256(report.read_bytes()).hexdigest()==receipt['report_sha256']; print('report hash matches receipt')" PATH-PRINTED-FOR-REPORT PATH-PRINTED-FOR-RECEIPT

Check the rejection path with a clearly synthetic structural fixture after you have a candidate. Copy its JSON to a separate file, remove any one required field from its first action, then invoke consume with that malformed file and a new empty output directory. It must fail before artifact creation. Never treat this fixture as a factual extraction or approve it. For example, the following local check removes evidence_quote from a copy:

Terminal
python -c "import json; from pathlib import Path; p=json.loads(Path('candidate/actions.json').read_text(encoding='utf-8')); p['actions'][0].pop('evidence_quote'); Path('candidate/SYNTHETIC-malformed.json').write_text(json.dumps(p), encoding='utf-8')"
python action_register_recipe.py consume --source-id OWNED-ID --source notes.txt --candidate candidate/SYNTHETIC-malformed.json --approval approval.json --out synthetic-rejection-output

Expect the second command to fail and the output directory to remain absent. Do not use the fixture if the real candidate has no first action; that candidate is already incomplete for this consumer. On an uncertain consumer write, inspect the deterministic report and receipt before retrying. Identical existing bytes can be reused; divergent bytes stop the operation for reconciliation. A changed source or schema version requires a fresh review. Never replay a model call merely because the local consumer response was unclear.

Complete Python recipe

Copy this script exactly into action_register_recipe.py:

Python
"""Private, bounded recipe. No model call occurs unless `produce` is explicitly run."""

import argparse
import hashlib
import json
import uuid
from datetime import date
from pathlib import Path
from typing import Literal

from pydantic import BaseModel, ConfigDict, StrictInt, StrictStr, ValidationError


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)


class Action(StrictModel):
    action_text: StrictStr
    evidence_quote: StrictStr
    line_start: StrictInt
    line_end: StrictInt
    owner: StrictStr | None
    due_date: StrictStr | None
    evidence_status: Literal["complete", "needs_review"]


class Extraction(StrictModel):
    schema_version: Literal["action-register/v1"]
    status: Literal["ready", "partial", "blocked"]
    source_id: StrictStr
    source_sha256: StrictStr
    actions: list[Action]
    gap_note: StrictStr | None


def digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def canonical(payload: Extraction) -> bytes:
    return json.dumps(payload.model_dump(mode="json"), sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def check(payload: Extraction, source_id: str, source: bytes) -> None:
    if payload.source_id != source_id or payload.source_sha256 != digest(source):
        raise ValueError("source identity or revision mismatch")
    lines = source.decode("utf-8").splitlines()
    if payload.status == "ready" and (not payload.actions or payload.gap_note is not None):
        raise ValueError("ready requires actions and no gap")
    if payload.status != "ready":
        raise ValueError("partial or blocked result cannot reach consumer")
    for index, item in enumerate(payload.actions, 1):
        if item.evidence_status != "complete":
            raise ValueError(f"action {index}: evidence needs review")
        if not item.action_text.strip() or not item.evidence_quote.strip():
            raise ValueError(f"action {index}: empty task or evidence")
        if not 1 <= item.line_start <= item.line_end <= len(lines):
            raise ValueError(f"action {index}: invalid source line range")
        segment = "\n".join(lines[item.line_start - 1:item.line_end])
        if item.evidence_quote not in segment:
            raise ValueError(f"action {index}: quotation absent from source span")
        if item.owner is not None and not item.owner.strip():
            raise ValueError(f"action {index}: blank owner")
        if item.due_date is not None:
            try:
                if date.fromisoformat(item.due_date).isoformat() != item.due_date:
                    raise ValueError()
            except ValueError as exc:
                raise ValueError(f"action {index}: due date must be YYYY-MM-DD") from exc


def report(payload: Extraction) -> str:
    """Application consumer: a triage register grouped by missing ownership/date."""
    groups = [
        ("Assigned with dates", lambda x: x.owner is not None and x.due_date is not None),
        ("Needs an owner", lambda x: x.owner is None),
        ("Needs a date", lambda x: x.owner is not None and x.due_date is None),
    ]
    rows = ["# Action register for " + payload.source_id, "", "Review artifact; source meaning was approved separately.", ""]
    for heading, predicate in groups:
        rows += ["## " + heading, ""]
        matches = [(n, a) for n, a in enumerate(payload.actions, 1) if predicate(a)]
        for number, action in matches:
            rows += [f"{number}. {action.action_text}", f"   Owner: {action.owner or 'Unspecified'}; due: {action.due_date or 'Unspecified'}", f"   Source lines {action.line_start}-{action.line_end}: {action.evidence_quote}", ""]
        if not matches:
            rows += ["None.", ""]
    return "\n".join(rows)


def load_payload(path: Path) -> Extraction:
    return Extraction.model_validate_json(path.read_bytes(), strict=True)


def consume(source_id: str, source_path: Path, payload_path: Path, approval_path: Path, out_dir: Path) -> tuple[Path, Path]:
    source = source_path.read_bytes()
    payload = load_payload(payload_path)
    check(payload, source_id, source)
    run = json.loads(payload_path.with_suffix(".run.json").read_text(encoding="utf-8"))
    payload_hash = digest(canonical(payload))
    if run.get("source_id") != source_id or run.get("source_sha256") != digest(source) or run.get("payload_sha256") != payload_hash or run.get("last_agent") != "Owned-source action extractor" or not str(run.get("run_id", "")).strip():
        raise ValueError("producer receipt does not match accepted candidate")
    approval = json.loads(approval_path.read_text(encoding="utf-8"))
    if (type(approval) is not dict
            or set(approval) != {"approved", "run_id", "source_id", "source_sha256", "payload_sha256", "reviewer", "review_note"}
            or approval["approved"] is not True
            or any(type(approval[key]) is not str for key in ("run_id", "source_id", "source_sha256", "payload_sha256", "reviewer", "review_note"))
            or not approval["reviewer"].strip() or not approval["review_note"].strip()
            or approval["run_id"] != run["run_id"]
            or approval["source_id"] != source_id
            or approval["source_sha256"] != digest(source)
            or approval["payload_sha256"] != payload_hash):
        raise ValueError("human source/meaning approval must bind exact source and payload hashes")
    body = report(payload).encode("utf-8")
    stem = "action-register-" + payload_hash[:16]
    out_dir.mkdir(parents=True, exist_ok=True)
    target = out_dir / (stem + ".md")
    receipt = out_dir / (stem + ".receipt.json")
    record = {"run_id": run["run_id"], "source_id": source_id, "source_sha256": digest(source), "payload_sha256": payload_hash, "report_sha256": digest(body), "reviewer": approval["reviewer"], "operation": "local action-triage register"}
    encoded = (json.dumps(record, indent=2) + "\n").encode("utf-8")
    for path, data in [(target, body), (receipt, encoded)]:
        if path.exists():
            if path.read_bytes() != data:
                raise ValueError(f"uncertain prior output differs: {path}; inspect before retry")
        else:
            with path.open("xb") as handle:
                handle.write(data)
        if path.read_bytes() != data:
            raise IOError(f"readback failed: {path}")
    return target, receipt


def produce(source_id: str, source_path: Path, candidate_path: Path, model: str) -> None:
    from agents import Agent, Runner

    source = source_path.read_bytes()
    source_text = source.decode("utf-8")
    if not source_text.strip() or candidate_path.exists() or not candidate_path.parent.is_dir():
        raise ValueError("source must contain text; candidate must be new and its parent directory must already exist")
    run_id = str(uuid.uuid4())
    agent = Agent(
        name="Owned-source action extractor",
        model=model,
        instructions=("Extract only explicit action commitments from the supplied numbered source. "
                      "Return action-register/v1. Copy an exact quote with matching 1-based line span. "
                      "Use null for an unstated owner or due date; never infer them. "
                      "Use partial and needs_review for ambiguous evidence; blocked for unusable input. "
                      "Do not use tools or hand off."),
        output_type=Extraction,
    )
    numbered = "\n".join(f"{n}: {line}" for n, line in enumerate(source_text.splitlines(), 1))
    try:
        result = Runner.run_sync(agent, f"Source ID: {source_id}\nSHA-256: {digest(source)}\nNumbered source:\n{numbered}")
        if result.interruptions or result.last_agent is not agent or not isinstance(result.final_output, Extraction):
            raise ValueError("run interrupted, unfinished, or finished by unexpected agent; this CLI does not persist resumable RunState, so do not resume from its failure receipt")
        output = result.final_output
        if output.source_id != source_id or output.source_sha256 != digest(source):
            raise ValueError("producer returned wrong source identity")
        candidate = canonical(output)
        with candidate_path.open("xb") as handle:
            handle.write(candidate)  # Candidate only; consumer still requires review.
        receipt = {"run_id": run_id, "source_id": source_id, "source_sha256": digest(source), "payload_sha256": digest(candidate), "last_agent": result.last_agent.name, "model": model}
        with candidate_path.with_suffix(".run.json").open("x", encoding="utf-8") as handle:
            handle.write(json.dumps(receipt, indent=2) + "\n")
    except Exception as exc:
        failure = {"run_id": run_id, "source_id": source_id, "source_sha256": digest(source), "error_type": type(exc).__name__, "error": str(exc), "state": "unknown_or_blocked; inspect run and destination before retry"}
        with candidate_path.with_name(candidate_path.stem + "." + run_id + ".failure.json").open("x", encoding="utf-8") as handle:
            handle.write(json.dumps(failure, indent=2) + "\n")
        raise


def main() -> None:
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    for name in ("produce", "consume"):
        p = sub.add_parser(name)
        p.add_argument("--source-id", required=True)
        p.add_argument("--source", type=Path, required=True)
        if name == "produce":
            p.add_argument("--candidate", type=Path, required=True)
            p.add_argument("--model", required=True)
        else:
            p.add_argument("--candidate", type=Path, required=True)
            p.add_argument("--approval", type=Path, required=True)
            p.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    if args.command == "produce":
        produce(args.source_id, args.source, args.candidate, args.model)
    else:
        print(*consume(args.source_id, args.source, args.candidate, args.approval, args.out), sep="\n")


if __name__ == "__main__":
    main()

Limits and sources

The script's contract and local consumer were checked offline with five tests and seven synthetic mocked producer or retry cases. Those checks exercised strict schema generation, rejection before artifact creation, grouped output, hashes and failure handling. They did not submit a schema to a provider, call a model, extract real notes, obtain factual human approval, create a real-source report or measure a bill or quality. Your API use has account-dependent charges; check current model compatibility, pricing and data permission before sending the source.

For adjacent context, BIG CHANGE's guide to choosing an AI task from owned work covers the earlier decision about which work to test. This guide begins with a selected, permitted extraction task and carries one typed result into a checked local consumer.

Sources & further reading