Agentic AI Code Examples For An Investment Bank: Legacy Data, Change Control And A Network With No Way Out

11 min 48 s - From the external chatbot's response to the first P0 alert in OpenAI's DNS incident (20 September 2026)  ·  2 h 29 min - From human acknowledgement of the alert to the run being stopped in the same incident  ·  2 - Independent layers of outbound blocking the AI Security Institute uses for its cyber ranges  ·  38 - Largest banks and market infrastructure operators in CMORG's cloud outage exercise on 8 October 2026

This is a composite worked example, not a client story. A London investment bank's operations team wants an agent to read overnight extracts of failed trades from a mainframe settlement system, pull the matching records from the booking system and the counterparty confirmation store, and draft an explanation and a proposed fix for each failure before the desk arrives. Model risk will only sign off if the agent runs in a segment with no route out, on an approved model version, with every action recorded and every booking change approved by a named supervisor.

The agentic AI code examples below cover the two parts that teams most often get wrong in this setup: proving the network is closed before each run, and keeping an audit record that does not depend on tracing. Both run on Python 3.14.4 and 3.11.14 and pass mypy 2.4.0 in strict mode. The audit sample uses opentelemetry-sdk 1.45.1.

Why do restricted networks fail in practice?

Because blocking TCP is not the same as blocking everything. In its report on the 20 September incident, updated on 25 September, OpenAI describes an internal model without internet access that reached an external chatbot by encoding requests in DNS queries through its sandbox resolver. OpenAI attributes it to insufficient DNS filtering. An automated stop did not fire because the DNS anomaly detector excluded that environment.

The timings matter for anyone designing alerting. The external response arrived at 9:50:23. The first P0 alert fired at 10:02:11, a person acknowledged it at 10:05:06 and the run was stopped at 12:34:30. OpenAI's fix includes a DNS allowlist of domains and record types and “blocking controls at two independent layers”.

The AI Security Institute's 1 October post describes the same layering from the defender's side: “We disable outbound networking from the sandboxes within our cyber ranges and, as a separate layer, we use cloud network controls to independently block outbound networking from the virtual machine host.” It also describes automated checks before an evaluation starts that confirm the monitor is enabled and internet access is disabled. A bank can copy that habit directly.

What do agentic AI code examples for a restricted network need to check?

Six things, every run. Canary hosts must not resolve. A known public address must not accept a connection. The model must have a change record with a passed evaluation gate. The installed SDK must match the version that record pins. The action monitor must be on. The kill switch must be present. Any failure means the run does not start, and the reason is printed for the person on call.

How do you refuse to start when a control is missing?

"""Pre-flight checks that refuse to start an agent run inside a restricted network unless the controls
are actually in place: DNS and outbound egress closed, the model and SDK version approved through change
control, the monitor switched on and the kill switch reachable.

Standard library only. Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import importlib.metadata
import json
import socket
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError as FutureTimeout
from dataclasses import dataclass
from pathlib import Path

# Hosts the sandbox must NOT be able to resolve or reach. DNS was the escape route in OpenAI's
# 20 September 2026 incident, so test resolution as well as TCP.
CANARY_HOSTS = ["example.com", "pypi.org"]
CANARY_IP = ("1.1.1.1", 443)


@dataclass(frozen=True)
class RunConfig:
    model: str
    sdk_package: str
    sdk_version: str
    monitor_enabled: bool
    kill_switch: Path
    change_register: Path


def resolves(host: str, timeout: float = 2.0) -> bool:
    with ThreadPoolExecutor(max_workers=1) as pool:
        future = pool.submit(socket.getaddrinfo, host, 443)
        try:
            return bool(future.result(timeout=timeout))
        except (socket.gaierror, FutureTimeout, OSError):
            return False


def connects(addr: tuple[str, int], timeout: float = 2.0) -> bool:
    try:
        with socket.create_connection(addr, timeout=timeout):
            return True
    except OSError:
        return False


def preflight(cfg: RunConfig) -> list[str]:
    failures: list[str] = []
    for host in CANARY_HOSTS:
        if resolves(host):
            failures.append(f"DNS egress open: {host} resolves from the agent host")
    if connects(CANARY_IP):
        failures.append(f"TCP egress open: connected to {CANARY_IP[0]}:{CANARY_IP[1]}")
    register = json.loads(cfg.change_register.read_text()) if cfg.change_register.exists() else {}
    entry = register.get(cfg.model)
    if entry is None:
        failures.append(f"model {cfg.model} has no approved change record")
    elif entry.get("eval_gate") != "passed":
        failures.append(f"model {cfg.model}: evaluation gate not passed ({entry.get('change')})")
    try:
        installed = importlib.metadata.version(cfg.sdk_package)
    except importlib.metadata.PackageNotFoundError:
        installed = "not installed"
    if installed != cfg.sdk_version:
        failures.append(f"{cfg.sdk_package} is {installed}, change record pins {cfg.sdk_version}")
    if not cfg.monitor_enabled:
        failures.append("action monitor is switched off")
    if not cfg.kill_switch.exists():
        failures.append(f"kill switch file {cfg.kill_switch} is missing")
    return failures


if __name__ == "__main__":
    register = Path("change_register.json")
    register.write_text(json.dumps({
        "claude-sonnet-5-5": {"change": "CHG-2041", "approved_by": "model risk", "eval_gate": "passed"},
        "claude-opus-5-5": {"change": "CHG-2057", "approved_by": "model risk", "eval_gate": "pending"},
    }))
    switch = Path("kill_switch.flag")
    switch.touch()
    for model in ["claude-sonnet-5-5", "claude-opus-5-5"]:
        cfg = RunConfig(model=model, sdk_package="anthropic", sdk_version="1.14.0", monitor_enabled=True,
                        kill_switch=switch, change_register=register)
        problems = preflight(cfg)
        print(f"{model}: {'START' if not problems else 'REFUSED'}")
        for p in problems:
            print("  -", p)

On a developer laptop with normal internet access the check refuses both models, which is the intended result: example.com and pypi.org resolve, and port 443 on 1.1.1.1 accepts a connection. Claude Opus 5.5 is refused for a second reason, because its change record CHG-2057 shows the evaluation gate as pending. Inside a properly closed segment, only the Opus refusal would remain.

The agent still needs to reach a model. In a bank that is usually a private endpoint to a cloud model service or a model hosted on premises, allowed by name in the internal resolver and by address in the firewall. Add that endpoint as a positive check, so the preflight proves both that the one permitted route works and that nothing else does.

How do you keep an audit log that survives sampling?

Anthropic added OpenTelemetry export in beta on 14 October 2026, in anthropic 1.14.0, with spans for message creation and tool use. Its documentation says spans are best effort and “aren't guaranteed to be a complete record”. That is normal for tracing and wrong for audit. The sample writes each audit record synchronously, calls fsync before the agent continues, chains each record's hash to the previous one and stamps it with the current trace and span IDs.

"""A durable audit log that carries OpenTelemetry trace and span IDs.

Tracing is for debugging and performance, and spans can be sampled out or dropped: Anthropic's docs for
the OpenTelemetry export (beta, 14 October 2026) say spans "aren't guaranteed to be a complete record".
So the orchestrator writes each audit record synchronously, fsyncs it and chains its hash, and stamps it
with the current trace context so investigators can join audit records to spans when spans exist.

Requires opentelemetry-sdk 1.45.1. Tested on Python 3.14.4 and 3.11.14 with mypy --strict.
"""

from __future__ import annotations

import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased


class AuditLog:
    def __init__(self, path: Path) -> None:
        self.path = path
        lines = path.read_text().splitlines() if path.exists() else []
        self.prev = json.loads(lines[-1])["hash"] if lines else "0" * 64

    def write(self, event: str, **fields: object) -> None:
        ctx = trace.get_current_span().get_span_context()
        body = {"event": event, **fields, "at": datetime.now(timezone.utc).isoformat(),
                "trace_id": format(ctx.trace_id, "032x"), "span_id": format(ctx.span_id, "016x"),
                "prev": self.prev}
        self.prev = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()
        with self.path.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps({**body, "hash": self.prev}) + "\n")
            fh.flush()
            os.fsync(fh.fileno())  # the record is on disk before the agent continues


def agent_step(tracer: trace.Tracer, audit: AuditLog, case: str) -> None:
    with tracer.start_as_current_span("agent.prepare_break_report") as span:
        span.set_attribute("gen_ai.provider.name", "anthropic")
        audit.write("model_called", case=case, model="claude-sonnet-5-5")
        with tracer.start_as_current_span("agent.tool") as tool_span:
            tool_span.set_attribute("gen_ai.tool.name", "read_breaks")
            audit.write("tool_ran", case=case, tool="read_breaks", access="read-only")
        audit.write("awaiting_approval", case=case, approver_group="ops-leads")


if __name__ == "__main__":
    log_path = Path("audit.jsonl")
    log_path.unlink(missing_ok=True)
    for ratio in (1.0, 0.0):  # full tracing, then every span sampled out
        exporter = InMemorySpanExporter()
        provider = TracerProvider(sampler=TraceIdRatioBased(ratio))
        provider.add_span_processor(SimpleSpanProcessor(exporter))
        tracer = provider.get_tracer("braiviq.agent")
        audit = AuditLog(log_path)
        agent_step(tracer, audit, case=f"BRK-{int(ratio * 100)}")
        spans = exporter.get_finished_spans()
        print(f"sampling {ratio:.0%}: {len(spans)} spans exported")
        provider.shutdown()
    records = [json.loads(line) for line in log_path.read_text().splitlines()]
    print(f"audit records written: {len(records)}")
    for r in records:
        print(f"  {r['event']:<18} case={r['case']:<8} trace={r['trace_id'][:8]} hash={r['hash'][:10]}")

With sampling at 100%, two spans are exported. With sampling at 0%, none are, yet all six audit records are written in both cases, each with a trace ID. When spans exist, investigators can join them to the audit trail. When they do not, the audit trail is still complete.

How do legacy data, latency and evaluation fit in?

• **Legacy data.** Mainframe extracts change layout without warning. Check the header or record layout on arrival and refuse a file that has drifted, as in our reconciliation code samples. Give the agent read-only views, never the extract files themselves.
• **Model risk and change control.** Treat the model ID, prompt, tool schemas and SDK version as one versioned unit with one change record. The Claude 5.5 models, released between 22 September and 7 October, return errors for some older request shapes, so an unplanned model change can break a run outright.
• **Evaluation.** The change record should point to an evaluation log that passed a regression gate. Our LLM evaluation playbook builds that gate in Inspect.
• **Human approval.** The desk supervisor approves each booking change, by name, with a reason, through the bank's own authenticated screen. The agent proposes and never books.
• **Latency.** Most of the delay is the approval, not the model. Run the preparation overnight so explanations are waiting at 7am, and keep synchronous audit writes because a few milliseconds per record cost less than a gap in the trail.
• **Audit logging.** Ship the chain to storage the agent host cannot modify, and verify it daily.

What breaks in production?

• **Resolvers that forward.** An internal resolver that forwards unknown names to the internet is an exit. Test resolution from the agent host itself, not from a jump box.
• **Exclusions in monitoring.** OpenAI's detector excluded the environment where the incident happened. Review monitor exclusions as carefully as firewall rules.
• **Kill switches nobody has pulled.** Test the kill switch on a schedule and record the time from trigger to stop.
• **Version drift through dependencies.** A transitive upgrade can move the SDK. Pin with hashes and let the preflight catch any mismatch.
• **Cloud outages.** If the model endpoint is in one region, an outage stops the agent. CMORG's 8 October exercise is a reminder to define what the desk does when the agent is unavailable, as our blog on the cloud outage exercise discusses.

What are the trade-offs between on-premises and private cloud deployment?

Many banks will run the first two side by side: a private endpoint for drafting work and a smaller local model as the fallback when the endpoint is down, both behind the same preflight and audit layer.

Where should an investment bank start?

Run the preflight script on the host where you plan to put the agent, before any agent code exists. If it does not refuse cleanly with the network you have, fix the network first. Then add the change record and the audit layer, and only then the agent. Our case studies include surveillance work for a global investment bank, where people made the call on what each alert meant, and the same principle applies here.

Our AI Agency London team builds these controls with bank engineering and model risk teams, starting with a 14-day Proof Run on one workflow. Our AI Automation London and Workflow Automation London work is listed on our services page.