Connectors, Contracts and the Human in the Loop

Module 5, Unit 2 | Lesson 3 of 3

By the end of this lesson, you will be able to:

• Design a connector interface using Python's abstract base class pattern, and implement concrete connectors for file and webhook outputs (K12, S10)
• Wire a connector into your output stage with a health check that halts the pipeline if the destination is unreachable (K11, K12)
• Implement a human approval checkpoint that pauses the pipeline for approval-required records before allowing processing to continue (K11, S10, B6)
• Explain why abstracted connectors and human checkpoints are not optional extras in a production AI pipeline (K11, B6)

Your pipeline can now ingest, validate, and route records through conditional branches with retry utilities ready for Unit 3. But a production pipeline does not exist in isolation — it reads from and writes to external systems, and it must hand certain decisions to a human when the stakes are too high for automation.

This lesson adds two architectural elements that complete Unit 2: a connector interface that abstracts your pipeline's output destinations, and a human governance checkpoint that enforces review before approval-required records are processed further.

What is a connector?

A connector is the code that sits at the boundary between your pipeline and an external system — a file store, a webhook, a database, a message queue. Its job is to translate between your pipeline's internal data format and whatever the external system expects.

Coach Cora

Good connector design means your pipeline logic never knows or cares what the external system looks like. If you switch from writing results to a local JSON file to posting them to a webhook, only the connector changes — not a single line of pipeline logic. This is the same boundary principle as the Pydantic schema in L1.2: the schema protects the pipeline from messy input; connectors protect it from messy output destinations.

Design the connector interface

Start with an abstract base class that defines the contract all connectors must fulfil. Create connectors.py:

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any


class OutputConnector(ABC):
    """
    Abstract base for all pipeline output connectors.
    Concrete implementations write to specific destinations.
    The pipeline logic depends only on this interface — not on any
    concrete implementation.
    """

    @abstractmethod
    def write(self, records: list[dict[str, Any]], metadata: dict[str, Any]) -> None:
        """Write a batch of records and pipeline metadata to the destination."""
        ...

    @abstractmethod
    def health_check(self) -> bool:
        """Return True if the destination is reachable and writable."""
        ...

The first implementation writes to a local file. serialise exists because date and Enum values are not JSON-serialisable, and that conversion belongs to the destination, not to the pipeline.

class JSONFileConnector(OutputConnector):
    """Writes pipeline output to a local JSON file."""

    def __init__(self, output_path: str) -> None:
        self.output_path = output_path

    def write(self, records: list[dict[str, Any]], metadata: dict[str, Any]) -> None:
        import json
        from datetime import date

        def serialise(obj: Any) -> Any:
            if isinstance(obj, date):
                return obj.isoformat()
            if hasattr(obj, "value"):
                return obj.value
            raise TypeError(f"Not serialisable: {type(obj)}")

        payload = {"metadata": metadata, "records": records}
        with open(self.output_path, "w", encoding="utf-8") as f:
            json.dump(payload, f, indent=2, default=serialise)
        print(f"[connector] {len(records)} records -> {self.output_path}")

    def health_check(self) -> bool:
        import os
        directory = os.path.dirname(self.output_path) or "."
        return os.access(directory, os.W_OK)

And here is why the interface was worth defining. A second destination, with nothing in common with the first but the two method names — and stage_output needs no change at all to use it.

class WebhookConnector(OutputConnector):
    """
    Sends pipeline output to an HTTP webhook endpoint.
    Replace url with a real endpoint (e.g. a Zapier webhook or internal API).
    """

    def __init__(self, url: str, api_key: str = "") -> None:
        self.url = url
        self.api_key = api_key

    def write(self, records: list[dict[str, Any]], metadata: dict[str, Any]) -> None:
        import json
        import urllib.request

        payload = json.dumps({"metadata": metadata, "records": records}).encode()
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        req = urllib.request.Request(self.url, data=payload, headers=headers)
        with urllib.request.urlopen(req, timeout=10) as response:
            print(f"[connector] webhook response: {response.status}")

    def health_check(self) -> bool:
        # Webhook endpoints only accept POST — a GET/HEAD probe would return 4xx
        # and falsely indicate the endpoint is unavailable. Return True and let
        # the actual write() call surface any real connectivity errors.
        return True

🔑 Key term — Abstract base class (ABC): A Python class that cannot be instantiated directly. It defines a set of methods that all subclasses must implement. ABCs are used to express interfaces — the contract that different implementations must fulfil. Code that depends on OutputConnector works with JSONFileConnector, WebhookConnector, or any future connector, without changing.

Curious Cat

Why use an abstract base class rather than two separate functions? Because the pipeline runner can accept any OutputConnector without knowing which one. In development you pass JSONFileConnector; in production you pass WebhookConnector. The pipeline code is identical. You can also write a test that passes a MockConnector that records what was written without touching the filesystem — which makes your pipeline stages fully unit-testable.

Wire the connector into the output stage

Two small helpers first, because they answer a question the stage itself should not have to: what exactly gets sent? The first flattens routed records into the payload shape. Add both to stages.py:

def output_records(state: PipelineState) -> list[dict[str, Any]]:
    """Flatten routed records into the shape a connector sends."""
    return [
        {
            "log_id": item["record"].log_id,
            "date": item["record"].date,
            "tool_name": item["record"].tool_name,
            "flagged_issue": item["record"].flagged_issue,
            "route": item.get("route", "unknown"),
            "classification": item.get("classification", "pending"),
            "cost_usd": item["record"].cost_usd,
            "human_reviewed": item["record"].human_reviewed,
        }
        for item in state.routed_records
    ]

The second describes the run rather than the records: which run this was, what it read, how each stage went, and whether it halted. It travels alongside the records so that whoever receives them can tell a complete run from a partial one.

Note that it carries each stage's errors list, not just a failure count. A count tells the recipient that three records did not make it; the list tells them which three and why. Those records are absent from the payload — a stage that fails a record drops it rather than passing a half-finished one downstream — so if the metadata does not name them, nothing does. A pipeline that quietly delivers 27 records when it read 30 is worse than one that fails loudly.

def run_metadata(state: PipelineState, record_count: int) -> dict[str, Any]:
    """The run's own story, sent alongside the records."""
    return {
        "run_id": state.run_id,
        "started_at": state.started_at.isoformat(),
        "source_file": state.source_file,
        "total_records": record_count,
        "halted": state.halted,
        "halt_reason": state.halt_reason,
        "stages": [
            {
                "stage": r.stage_name,
                "records_in": r.records_in,
                "records_out": r.records_out,
                "records_failed": r.records_failed,
                # Not just how many failed — which ones, and why. Without this
                # a dropped record leaves no trace in the delivered payload.
                "errors": r.errors,
                "duration_s": r.duration_seconds,
                "metadata": r.metadata,
            }
            for r in state.stage_results
        ],
    }

Now stage_output itself. It takes a connector, checks the destination is reachable before writing anything, and otherwise just shapes and hands over. Replace the L1.3 version in stages.py:

from connectors import OutputConnector, JSONFileConnector


def stage_output(
    state: PipelineState,
    connector: OutputConnector | None = None,
) -> PipelineState:
    """Stage 5: Write pipeline results via the configured connector."""
    started = time.time()
    if connector is None:
        connector = JSONFileConnector("pipeline_output.json")

    if not connector.health_check():
        state.halted = True
        state.halt_reason = (
            f"Output connector health check failed: "
            f"{connector.__class__.__name__}"
        )
        state.stage_results.append(StageResult(
            stage_name="output",
            records_in=len(state.routed_records),
            records_out=0,
            records_failed=len(state.routed_records),
            duration_seconds=round(time.time() - started, 3),
            metadata={
                "connector": connector.__class__.__name__,
                "health_check": "failed",
            },
        ))
        print("[output]   HALTING - connector unavailable")
        return state

    records = output_records(state)
    state.stage_results.append(StageResult(
        stage_name="output",
        records_in=len(state.routed_records),
        records_out=len(records),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={"connector": connector.__class__.__name__},
    ))
    connector.write(records, run_metadata(state, len(records)))
    return state

The health check earning its place is the point of that middle block: a destination that cannot be reached is a reason to stop, not to retry blindly into it. Everything after it is bookkeeping — and notice the stage never learns whether the result landed in a file or a webhook.

Update run_pipeline in pipeline.py to build a connector and pass it in. Two lines differ from L1.3 — the default connector, and the if not state.halted that now guards the write:

def run_pipeline(source_file: str, connector=None) -> PipelineState:
    if connector is None:
        connector = JSONFileConnector("pipeline_output.json")

    state = PipelineState(
        run_id=str(uuid.uuid4()),
        started_at=datetime.now(timezone.utc),
        source_file=source_file,
    )

    print(f"\n{'='*50}")
    print(f"Pipeline run: {state.run_id}")
    print(f"Source:       {state.source_file}")
    print(f"{'='*50}\n")

    for stage_fn in PROCESSING_STAGES:
        state = stage_fn(state)
        if state.halted:
            print(f"\nPipeline halted after [{stage_fn.__name__}].")
            break

    if not state.halted:
        state = stage_output(state, connector=connector)

    total_time = sum(r.duration_seconds for r in state.stage_results)
    print(f"\n{'='*50}")
    print(f"Completed in {total_time:.3f}s  |  Valid: {len(state.valid_records)}")
    print(f"{'='*50}\n")

    return state


if __name__ == "__main__":
    run_pipeline("ai_tool_usage_log.csv")

⚠️ Behaviour change since L1.3.

What changed. In L1.3 the runner called stage_output unconditionally, so a halted run still wrote pipeline_output.json with "halted": true. From this lesson on, output is guarded by if not state.halted, and a halted run writes no output file at all.

Why. The destination is now an external system rather than a local file. A failed health check or a rejected approval means that system must not receive a partial result.

What follows. Halt evidence now lives in the terminal, and from L3.4 in the structured log, rather than in pipeline_output.json. The L1.3 halt test no longer applies as written.

Add a human approval checkpoint

Some records must not be processed automatically. The approval checkpoint handles detected PII in the compliance queue, and it also handles any later record with route_override == "human_review". A human should verify these records before they go anywhere.

A human-in-the-loop checkpoint pauses the pipeline, presents the records requiring review, and waits for explicit approval. Three functions, added to stages.py. The first is only presentation — what the operator actually sees:

def show_approval_items(items: list[dict[str, Any]]) -> None:
    """Print the records an operator is being asked to approve."""
    print(f"\n{'='*50}")
    print("HUMAN APPROVAL REQUIRED")
    print(f"{len(items)} record(s) requiring approval:")
    print(f"{'='*50}")
    for item in items:
        r = item["record"]
        print(f"  {r.log_id} | {r.learner_id} | {r.tool_name.value} | {r.date}")
    print(f"{'='*50}")

The second collects the decision. input() blocks the pipeline, which is exactly the point of a checkpoint — but it also makes the stage unrunnable in CI, hence the explicit opt-out. It returns two answers: whether the records were approved, and whether a human was involved at all.

def collect_approval() -> tuple[bool, bool]:
    """Returns (approved, auto_approved)."""
    import os

    # Escape hatch for automated runs (benchmarks, CI). Off unless explicitly set.
    if os.environ.get("AUTO_APPROVE") == "1":
        print("[approval] AUTO_APPROVE=1 - approving without prompting.")
        return True, True

    response = input("\nApprove processing these records? [yes/no]: ").strip().lower()
    return response in ("yes", "y"), False

Now the stage. It selects the records that need approving, returns early if there are none, and records the decision either way. Note that a rejection halts rather than raises: a refused approval is a legitimate outcome of a run, not a crash.

def stage_human_approval(state: PipelineState) -> PipelineState:
    """Checkpoint: present approval-required records for human review."""
    started = time.time()

    approval_items = [
        item for item in state.routed_records
        if item.get("route") in ("compliance", "human_review")
    ]

    if not approval_items:
        print("[approval] No approval-required records - checkpoint skipped")
        state.stage_results.append(StageResult(
            stage_name="human_approval",
            records_in=len(state.routed_records),
            records_out=len(state.routed_records),
            records_failed=0,
            duration_seconds=round(time.time() - started, 3),
            metadata={"skipped": True, "reason": "no approval-required records"},
        ))
        return state

    show_approval_items(approval_items)
    approved, auto_approved = collect_approval()

    if not approved:
        state.halted = True
        state.halt_reason = (
            "Human operator rejected approval-required records at approval checkpoint"
        )
        print("[approval] Pipeline halted by operator.")
    else:
        print("[approval] Approved. Continuing pipeline.")

    state.stage_results.append(StageResult(
        stage_name="human_approval",
        records_in=len(state.routed_records),
        records_out=len(state.routed_records) if approved else 0,
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={
            "approval_records": len(approval_items),
            "approved": approved,
            "auto_approved": auto_approved,
        },
    ))

    return state

Two details are worth pausing on. The early return still appends a StageResult — a checkpoint that never fires has to leave evidence that it ran, otherwise "no PII today" and "the checkpoint was never wired in" look identical in the output. And approved and auto_approved both land in the stage metadata, so an auditor can tell a human decision from an automated one.

In production you would replace input() with a webhook that triggers an asynchronous approval flow — Slack, email, an internal tool. The structure of the stage is identical; only the mechanism for collecting the human decision changes.

Curious Cat

Does an AUTO_APPROVE flag not defeat the entire point of a governance checkpoint? It would — if it were silent. You will need it: from Unit 3 onwards you re-run this pipeline constantly, and the benchmark in L4.2 runs it four times in a row. Typing yes each time is friction with no learning in it. What makes the flag acceptable is "auto_approved": true landing in the stage metadata and therefore in pipeline_output.json. An auditor can see instantly that no human looked at those records. A bypass you can detect is an engineering convenience; a bypass you cannot detect is a control that does not exist. Use it for benchmarks, never for the run you submit as evidence.

Add it to PROCESSING_STAGES in pipeline.py, between route and output. Here is the whole header of the file as it now stands:

from __future__ import annotations

import uuid
from datetime import datetime, timezone

from connectors import JSONFileConnector
from contracts import PipelineState
from stages import (
    stage_ingest,
    stage_validate,
    stage_classify,
    stage_route,
    stage_human_approval,
    stage_output,
)

PROCESSING_STAGES = [
    stage_ingest,
    stage_validate,
    stage_classify,
    stage_route,
    stage_human_approval,
]

stage_output stays out of the list — it is called separately after the loop so the runner can inject the connector. That is the same split you set up in L1.3; the only change is that stage_human_approval now sits at the end of the processing stages.

Coach Cora

The input() call here is a deliberate simplification. In a real deployment you would replace it with a notification — a Slack message with Approve/Reject buttons, an email with a secure link, or a task in a case management system. The pipeline stage structure is identical in all of these cases; only the mechanism for collecting the human decision changes. That separation is the point: your pipeline does not need to know how the human reviews — it just needs to know whether they approved.

Build activity

1. Create connectors.py with OutputConnector, JSONFileConnector, and WebhookConnector.
2. Update stage_output in stages.py to accept and use a connector with a health check.
3. Update pipeline.py to pass JSONFileConnector("pipeline_output.json") explicitly.
4. Add stage_human_approval to stages.py and add it to the PROCESSING_STAGES list in pipeline.py.
5. Run python pipeline.py. The pipeline should pause at the approval checkpoint, display the PII-flagged records, and wait for your input. Type yes to continue to completion.

Verify that pipeline_output.json includes human_approval in the stage list, and that its metadata shows "approved": true and the correct count of approval-required records.

Challenge Chase

Take this to your own organisation. Where does data your team produces actually need to go — a warehouse, a CRM, a shared drive, a Teams channel, someone's inbox? Pick one and sketch the OutputConnector for it: what does write() send, and what would a meaningful health_check() actually check? Then the harder question: which decisions in that flow currently get a human sign-off, and is that sign-off enforced by a system or by someone remembering? A checkpoint that depends on memory is the one worth building into the pipeline.

The code from this lesson

The whole of Unit 2, ready to run: connectors, the approval checkpoint, and a runner that passes the connector in. Set AUTO_APPROVE=1 to run it without being prompted. The code files are assembled from the code blocks of this lesson and the ones before it.

Take it if you would rather read and run the code than type it out. Typing it is still the better way to learn it — this is here so a typo cannot cost you the lesson.

Checklist

• connectors.py exists and defines OutputConnector (ABC), JSONFileConnector and WebhookConnector
• stage_output accepts an OutputConnector parameter and runs a health check before writing
• pipeline.py passes JSONFileConnector("pipeline_output.json") explicitly to stage_output
• stage_human_approval is in stages.py and added to PROCESSING_STAGES in pipeline.py
• Running python pipeline.py pauses at the approval checkpoint when approval-required records are present
• After typing yes, the pipeline completes and pipeline_output.json shows "approved": true in the human_approval stage metadata
• I can explain the difference between input() as a development checkpoint and an async approval flow in production

Quick Check

KSB evidence focus

• K11 — Understands relevant data governance, data privacy and security issues. The human approval checkpoint is a governance control — it enforces human review of PII-flagged records before they are processed further. Building it into the pipeline architecture means the governance requirement is not an afterthought or a manual process check; it is a technical enforcement mechanism that cannot be bypassed.

• K12 — Understands how to set up, interact with and generate APIs, databases and spreadsheets. The WebhookConnector is a working HTTP client that sends structured JSON to an external endpoint. The abstract base class pattern means adding a new connector — a database writer, an S3 uploader, a message queue producer — requires only implementing two methods: write and health_check.

• B6 — Shows curiosity and initiative. You built the human approval checkpoint before it was required by the data. That is forward thinking — compliance requirements do not announce themselves. Anticipating the need and building the mechanism now, rather than retrofitting it after a PII incident, is the habit that separates a reactive developer from a proactive one.

Up next: Unit 3 replaces the agent stubs with real LLM logic. You will use the OpenAI API to classify flagged records, run Qwen3 1.7B locally with Ollama for sensitive data that cannot leave your environment, add structured logging to make every pipeline decision traceable, and build debugging tools for when things go wrong.