Part 3: how to prove, before every release and in production, that a governed data API gives people and AI agents the right answer, for the right person, in a form an agent can use
Press enter or click to view image in full size
Core definition. A data API is ready for AI callers when five things are proven automatically before every release and watched in production: the contract holds, the answer is right on every path, access is right for every user, the agent surface is usable, and nothing is drifting.
Part 1, “Data Is Becoming an API, and AI Is the New Caller”, argued that AI agents are becoming a major caller of enterprise data. Part 2, “Building an AI-Ready Data Services API on Azure, End to End”, built one: a governed net-revenue metric served over REST and as an MCP tool, with Microsoft Entra ID, budgets and tracing. This part answers the question an auditor, a risk officer or an on-call engineer will ask next: how do you know it keeps working?
Why this needs its own discipline. A person who gets a wrong number usually notices. An agent may pass it on in every answer it gives, and nobody may check. Unit tests written from a few examples do not cover what an agent will try: odd date ranges, regions it should not see, the same failing call again and again. The checks below are designed for that caller.
What was run. All Python ran locally against Part 2’s service, with DuckDB standing in for the Azure data store and test tokens signed with a local key. The main suite of 60 tests passed (Part 2’s 28 and 32 new), and the schema-generated contract suite passed with 100 generated requests. The release gate ran locally up to the deploy command. The Application Insights queries and the deploy command were checked against Microsoft’s documentation, not run.
Key takeaways
Let the schema write tests. On its first run, the schema-generated suite found two places where the published contract and the service disagreed.
Pin the answer, then prove every path agrees. The REST endpoint and the MCP tool must return the same rows, definition, source and freshness.
Test access as a matrix. Every user against every region, for the app and for an agent acting for that user.
Lint what the agent reads, and evaluate agents from their traces. Descriptions, schemas and error text are an interface; outcomes in the audit trail show which agents are struggling.
Make drift a release blocker. A breaking contract change or late data stops the deploy.
The assurance map
Press enter or click to view image in full size
Figure 1. Five properties of an AI-ready data API, and the checks that prove each one before merge, at the release gate and in production. Shaded parts are the checks that block a release.
1. Let the schema write the tests
Core: the published contract is a test generator. Use it, and treat every disagreement as a finding.
What: property-based contract testing. Schemathesis reads the OpenAPI document, generates many valid and invalid requests, calls the app in process and checks each response against the schema.
Why needed: an agent will send requests no one thought to write down. Hand-written tests confirm the cases the author imagined; generated tests explore the rest of what the contract allows.
How: load the schema from the ASGI app and teach the generator the one rule OpenAPI cannot express: a range of one to 24 months across two separate parameters.
"""Contract: let the published schema generate the requests, and check every response against it.""" import re import pytest import schemathesis import app as app_module from conftest import RAJ, make_token from models import MONTH LAST = 2099 * 12 + 11 # December 2099, the last month the published pattern allows @schemathesis.hook def map_query(ctx, query): """Teach the generator the one rule the schema cannot express: a range of 1 to 24 months.""" months = [query.get(k) for k in ("from_month", "to_month")] if isinstance(query, dict) else [] if len(months) == 2 and all(isinstance(v, str) and re.fullmatch(MONTH, v) for v in months): # valid cases only y, m = map(int, query["from_month"].split("-")) span = int(query["to_month"][-2:]) % 24 # 0-23 extra months, varied by the generator start = min(y * 12 + m - 1, LAST - span) to = start + span query["from_month"] = f"{start // 12}-{start % 12 + 1:02d}" query["to_month"] = f"{to // 12}-{to % 12 + 1:02d}" if "page_token" in query: # a page token must fall inside the range query["page_token"] = query["from_month"] return query @pytest.fixture def api_schema(): return schemathesis.openapi.from_asgi("/openapi.json", app_module.app) schema = schemathesis.pytest.from_fixture("api_schema") # loaded when the test runs, not at import @schema.parametrize() def test_every_generated_request_gets_a_documented_answer(case): case.call_and_validate(headers={"Authorization": f"Bearer {make_token(oid=RAJ)}"})
Effectiveness: the first run failed, correctly. Schemathesis generated from_month=0000-01, which Part 2's pattern allowed, and the service refused it with 422. Its report read: ST202 Schema-valid request rejected. The contract said yes and the service said no. Two fixes followed. The month pattern now allows only years 20xx. And the refusals that depend on the data rather than on the request's form now use the status codes that mean "a valid request, but not this data": 409 Conflict for a month that is not certified yet (RFC 9110 describes 409 as "a conflict with the current state of the target resource") and 404 Not Found for a month before the data begins. Both are declared in the published responses. The next runs caught a mistake I introduced while making that change, then found the cross-parameter rule, which is why the hook above exists. With both changes, 100 generated requests pass.
When applicable: any API with a machine-readable schema. Run it in its own process: Schemathesis starts the app’s lifespan, and the MCP session manager in Part 2’s app can start only once per process.
The two contract fixes, as excerpts from models.py and service.py:
# models.py MONTH = r"^20[0-9]{2}-(0[1-9]|1[0-2])$" # this data product starts in 2025 # service.py (messages shortened here) raise DataApiError(409, "month_not_closed", ...) # valid request; the month is not certified yet raise DataApiError(404, "before_first_month", ...) # valid request; no data that early
2. Pin the answer, then prove every path agrees
Core: a correct number on one path is not enough; every caller must get the same answer with the same evidence.
What: a golden test that pins certified values, and a parity test that compares the REST answer with the MCP tool’s answer.
Why needed: the agent path and the app path can drift apart quietly, through a different default, a different rounding or a missing field. An agent that reports a different number from the dashboard destroys trust in both.
How:
GOLDEN = {("WEST", "2026-09"): 1840210.55, ("WEST", "2026-10"): 1902774.10} def test_golden_answers(app): body = TestClient(app).get(URL, params=PART1, headers={"Authorization": f"Bearer {make_token()}"}).json() assert {("WEST", r["month"]): r["net_revenue_usd"] for r in body["rows"]} == GOLDEN def test_same_answer_on_every_path(live_server): import httpx with httpx.Client(base_url=live_server, headers={"Authorization": f"Bearer {make_token()}"}) as http: rest = fetch_net_revenue(http, **PART1) ok, mcp_answer = anyio.run(call_net_revenue, make_token(azp=AGENT_APP), PART1, f"{live_server}/agents/mcp") assert ok for key in ("rows", "metric_definition", "as_of", "source"): assert rest[key] == mcp_answer[key], key
Effectiveness: both tests passed over real HTTP. The parity test compares everything except the trace ID, which differs by design.
When applicable: whenever one metric is served through more than one path. Refresh the golden values only through the same review that certifies the data.
3. Test access as a matrix, not as examples
Core: for an agent acting on someone’s behalf, every user and every object is a test case.
What: one parameterized test that runs every user against every region, once through the app and once through an agent acting for that user.
Why needed: the OWASP API Security Top 10 (2023) ranks Broken Object Level Authorization first. A handful of example tests misses the combinations, and agents try combinations.
How:
USERS = {ANA: {"WEST"}, RAJ: set(REGIONS), "oid-no-entitlements": set()} @pytest.mark.parametrize("oid,region,client_app", list(product(USERS, REGIONS, [WEB_APP, AGENT_APP]))) def test_access_matrix(app, oid, region, client_app): r = TestClient(app).get(URL, params={**PART1, "region": region}, headers={"Authorization": f"Bearer {make_token(oid=oid, azp=client_app)}"}) if region in USERS[oid]: assert r.status_code == 200 and {row["month"] for row in r.json()["rows"]} == {"2026-09", "2026-10"} else: assert r.status_code == 403 and r.json()["code"] == "region_not_entitled"
Effectiveness: 24 cases (3 users, 4 regions, 2 calling apps), all passing. The agent never received more than its user may see.
When applicable: any API that filters by owner, region, account or tenant. Generate the matrix from the entitlement table, so new users and regions are covered without new code.
4. Lint what the agent reads
Core: for an AI caller, the tool description, schema and error text are the interface, so check them like code.
What: a linter for the MCP tool definition and for the error text a model receives.
Why needed: a vague description or an unbounded input lets a model call the tool for the wrong question or with arguments it cannot validate. An error without a next step leaves the model to guess.
How:
"""Lint the tool surface an AI agent reads: description, input bounds, output contract, annotations.""" import re REQUIRED_EVIDENCE = {"metric_definition", "as_of", "source", "trace_id"} def _bounded(prop: dict) -> bool: return any(k in prop for k in ("enum", "pattern", "maximum", "maxLength", "const")) RULES = { "description_states_scope": lambda t: len(t.description or "") >= 80, "description_says_what_it_does_not_do": lambda t: bool(re.search(r"\b(does not|do not)\b", t.description or "", re.I)), "every_input_is_bounded": lambda t: all(_bounded(p) for p in t.input_schema.get("properties", {}).values()), "inputs_are_required_or_defaulted": lambda t: all( name in t.input_schema.get("required", []) or "default" in p for name, p in t.input_schema.get("properties", {}).items()), "output_carries_its_evidence": lambda t: REQUIRED_EVIDENCE <= set((t.output_schema or {}).get("required", [])), "declares_read_only": lambda t: bool(t.annotations and t.annotations.read_only_hint), } def lint_tool(tool) -> list[str]: """Return the names of the rules this tool fails; an empty list means it passes.""" return [name for name, rule in RULES.items() if not rule(tool)] def lint_error_text(text: str) -> list[str]: """An error the model will read must say what was wrong and what to do instead.""" problems = [] if len(text) < 40: problems.append("error_too_short_to_act_on") if not re.search(r"\b(retry|use|ask|send|call)\b", text, re.I): problems.append("error_gives_no_next_step") return problems
Effectiveness: Part 2’s tool passed every rule, and so did its error text for an open month. As a control, the generic message “Invalid request.” failed both error rules. The read-only declaration is a hint for clients, not a control: the MCP specification says clients “MUST consider tool annotations to be untrusted unless they come from trusted servers,” so the server still enforces access itself (section 3).
When applicable: every tool or endpoint an agent can call. Agree the rules with the teams that write tools, and keep them few enough to remember.
5. Evaluate agents from their traces
Core: the audit trail already records how each agent uses the data; score it.
What: an evaluator that reads the data_api.* attributes on each request span and reports, per calling app or agent, its success rate, how often it recovers after an error, how often it repeats the same error and how often it is refused.
Why needed: a model that ignores error text, or keeps asking for data its user may not see, will not show up in unit tests. It shows up in behaviour. You can measure that behaviour without access to the model.
How: one change to Part 2’s service records the outcome on every answer and every refusal, on both paths. For example, the REST error handler gains audit(outcome=e.code) and a successful answer records outcome="ok". The evaluator then reads the spans:
"""Evaluate how each calling app or agent uses the data API, from the audit attributes on its spans.""" from collections import defaultdict def calls_from_spans(spans) -> list[dict]: """One record per answered or refused call, in time order, from spans that carry an outcome.""" calls = [] for s in spans: a = dict(s.attributes or {}) if "data_api.outcome" in a: calls.append({"time": s.start_time, "trace": f"{s.context.trace_id:032x}", "app": a.get("data_api.client_app"), "user": a.get("data_api.user_oid"), "path": a.get("data_api.path"), "outcome": a["data_api.outcome"]}) return sorted(calls, key=lambda c: c["time"]) def evaluate(calls: list[dict], window: int = 2) -> dict[str, dict]: """Per calling app: success rate, refusals, recovery after an error, and repeated identical errors.""" by_app = defaultdict(list) for c in calls: by_app[c["app"]].append(c) report = {} for app, seq in by_app.items(): errors = [i for i, c in enumerate(seq) if c["outcome"] != "ok"] recovered = sum(any(n["outcome"] == "ok" for n in seq[i + 1:i + 1 + window]) for i in errors) repeats = sum(1 for a, b in zip(seq, seq[1:]) if a["outcome"] != "ok" and a["outcome"] == b["outcome"]) outcomes: dict[str, int] = {} for c in seq: outcomes[c["outcome"]] = outcomes.get(c["outcome"], 0) + 1 report[app] = { "calls": len(seq), "success_rate": round(outcomes.get("ok", 0) / len(seq), 2), "recovery_rate": round(recovered / len(errors), 2) if errors else None, "repeated_errors": repeats, "entitlement_refusals": outcomes.get("region_not_entitled", 0), "outcomes": outcomes, } return report def verdict(r: dict, min_recovery: float = 0.8) -> str: """A simple gate: an agent that repeats the same error, or rarely recovers, needs attention.""" if r["repeated_errors"] or (r["recovery_rate"] is not None and r["recovery_rate"] < min_recovery): return "review" return "pass"
Effectiveness: two scripted callers, standing in for model-driven agents, made real calls over Streamable HTTP. The “careful” caller asked for an open month, read the error and corrected itself: success rate 0.5, recovery rate 1.0, verdict pass. The "sloppy" caller asked for two open months in a row and then for a region its user may not see: one repeated error, one entitlement refusal, verdict review. The evaluator scores behaviour, not the model's reasoning; it tells you which agent to examine, not why it failed.
When applicable: as soon as more than one agent or app calls the API. In production, the same attributes are in Application Insights (section 7).
6. Make drift a release blocker
Core: two kinds of drift break AI callers quietly: a contract change, and data that is late or out of pattern.
What: a contract diff that lists changes that can break an existing caller, and data checks for freshness and level shifts.
Why needed: an agent built against last month’s schema will not read release notes. And a certified table that stops updating keeps answering, with a correct-looking as_of that is simply old.
How: an excerpt of drift.py:
def breaking_changes(old: dict, new: dict) -> list[str]: """Changes that can break an existing caller of the API (additions are allowed).""" found = [] for path, ops in old["paths"].items(): for method, op in ops.items(): new_op = new["paths"].get(path, {}).get(method) if new_op is None: found.append(f"{method.upper()} {path}: operation removed") continue old_params = {p["name"]: p for p in op.get("parameters", [])} for p in new_op.get("parameters", []): was = old_params.get(p["name"]) if p.get("required") and not (was and was.get("required")): found.append(f"{path}: parameter '{p['name']}' is now required") if was and "enum" in was.get("schema", {}): lost = set(was["schema"]["enum"]) - set(p.get("schema", {}).get("enum", was["schema"]["enum"])) if lost: found.append(f"{path}: parameter '{p['name']}' no longer accepts {sorted(lost)}") for name in old["components"]["schemas"]: if name not in new["components"]["schemas"]: found.append(f"schema {name} removed") continue (old_p, old_req), (new_p, new_req) = _props(old, name), _props(new, name) for prop in old_p: if prop not in new_p: found.append(f"{name}.{prop} removed") elif old_p[prop].get("type") != new_p[prop].get("type"): found.append(f"{name}.{prop} changed type") for prop in old_req - new_req: if prop in new_p: found.append(f"{name}.{prop} is no longer guaranteed") return found
def data_checks(con, today: date, sla_days: int = 3, mad_limit: float = 5.0) -> list[str]: """Late data (the contract's latency SLA) and month values far outside each region's recent pattern.""" found = [] y, m = today.year, today.month - (1 if today.day > sla_days else 2) # last month that should be certified while m < 1: y, m = y - 1, m + 12 expected = f"{y}-{m:02d}" latest = con.execute("SELECT MAX(month) FROM finance.net_revenue_monthly").fetchone()[0] if latest < expected: found.append(f"stale: latest certified month is {latest}, expected {expected} within {sla_days} days") rows = con.execute("""SELECT region, month, gross_sales_usd - returns_usd - discounts_usd FROM finance.net_revenue_monthly ORDER BY region, month""").fetchall() series: dict[str, list[tuple[str, float]]] = {} for region, month, value in rows: series.setdefault(region, []).append((month, float(value))) for region, points in series.items(): *history, (month, value) = points[-7:] values = [v for _, v in history] median = statistics.median(values) mad = statistics.median(abs(v - median) for v in values) or 1.0 if abs(value - median) / mad > mad_limit: found.append(f"{region} {month}: {value:,.2f} is far outside the last six months (median {median:,.2f})") return found
Effectiveness: adding an optional currency field was correctly allowed. Removing source, making limit required and dropping EAST from the region list were all caught, by name. On the data side, the table passed on November 5; on December 4 it was flagged as stale against the three-day latency in the data contract; and a planted tripling of one region's month was flagged as far outside its last six months. The level-shift check uses the median and the median absolute deviation, which a single earlier outlier cannot drag around as it would a mean.
When applicable: run the contract diff on every change to the API, and the data checks on a schedule and before every release. Set the deviation limit from your own history; the value here is an example.
7. In production: the same signals, as queries
Core: what the tests prove before release, telemetry should keep showing after it.
What: two Kusto queries over the Application Insights AppRequests table.
Why needed: traffic in production is the only place where real agents, real users and real data meet.
How: both paths record the outcome on a request (server) span, which the Azure Monitor exporter sends as a request with the span’s attributes as custom properties, so they appear in the Properties column:
// Outcomes by calling app and agent over the last day; span attributes arrive in Properties AppRequests | where TimeGenerated > ago(1d) | extend client_app = tostring(Properties["data_api.client_app"]), outcome = tostring(Properties["data_api.outcome"]) | where isnotempty(outcome) | summarize calls = count() by client_app, outcome | order by client_app asc, calls desc // Reconstruct any answer from the trace_id it returned AppRequests | where OperationId == "<trace_id from the response>" | project TimeGenerated, Name, ResultCode, DurationMs, Properties
Effectiveness: not run here; the table and column names follow Microsoft’s AppRequests reference. The first query feeds the same evaluation as section 5; the second turns any disputed answer into its request, user, calling app and data version.
When applicable: daily review of agent behaviour, and any time an answer is questioned.
8. One gate before every release
Core: every check above runs before a deploy, and any failure stops it.
#!/usr/bin/env bash # Release gate for the data API: run in CI before every deploy. Any failing step stops the release. set -euo pipefail pytest -q tests --ignore=tests/test_schema_generated.py # answers, access, agent surface, traces, data drift pytest -q tests/test_schema_generated.py # schema-generated requests, in their own process python -c "import app, json; json.dump(app.app.openapi(), open('current_openapi.json', 'w'))" python drift.py baseline/openapi.json current_openapi.json # exits 1 on any breaking contract change # Only after every check passes: build and deploy (Part 2, section 9) az containerapp up --name net-revenue-api --resource-group rg-net-revenue --source . \ --ingress external --target-port 8000
Locally, every step before the deploy command passed, and the contract diff returned exit code 1 when fed a contract with source removed, which would have stopped the release.
What each check proves
Press enter or click to view image in full size
Figure 2. Each check, what it proves, why it is needed, the Python used, when it runs and how effective it was here.
What these checks do not prove
Core: know the edges of your evidence.
Model reasoning. The trace evaluation scores what an agent did with the data, not whether its final answer to a person was right. That needs a separate evaluation of the agent itself.
Real identity and gateways. Tokens were signed locally; Entra-issued tokens and API Management policies were not exercised.
Scale. None of this measures behaviour under production load or concurrency.
The data’s truth. The golden values prove the API returns what was certified, not that the certification was correct.
A checklist: is your data API proven for AI callers?
Contract: does a schema-generated suite run on every change, in its own step?
Answer: are certified values pinned, and do all paths return the same answer and evidence?
Access: is every user tested against every object, for apps and for agents acting for them?
Agent surface: are tool descriptions, input bounds and error text linted?
Behaviour: is every call’s outcome in the audit trail, and is each agent’s recovery rate reviewed?
Drift: do breaking contract changes and late or out-of-pattern data block a release?
Production: can any answer be traced back to its request, user, calling app and data version?
The core, in one line
A data API is ready for AI callers when its contract, answers, access, agent surface and freshness are proven by code before every release and kept visible in production.
Get Reno Christy’s stories in your inbox
Join Medium for free to get updates from this writer.
I used an AI writing assistant to help draft and edit this article. I reviewed and revised the text and checked every source.
