refactor: FastAPI migration — unify MCP + A2A + Admin into single process

[KonstantinMirin]

Closes #1050

Summary

Unify the three separate processes (MCP, A2A, Flask Admin) into a single FastAPI process with clean separation of concerns — and fix the test suite from 138 warnings / flaky failures to 3,490 passing tests, 0 warnings, 51% line coverage.

The core problem

Every transport (MCP, A2A, Admin) reimplemented auth extraction, tenant resolution, error formatting, and serialization. Every _impl() function reached into framework concerns — reading tenant from ContextVars, calling set_current_tenant(), catching transport-specific ToolError. This made it impossible to add a transport without duplicating all of that, and impossible to change any of it without touching every transport.

What changed: data flow unification

The key idea is that each concern is handled exactly once, in one place.

Before: every layer does everything

Each handler and _impl() function mixes auth, tenant resolution, error handling, serialization, and version compat. Adding a new tool means copying all of it.

flowchart TD
    subgraph MCP["MCP Handler"]
        M1[Extract auth token]
        M2[Call _impl]
        M3[set_current_tenant]
        M4[get_current_tenant from ContextVar]
        M5[get_adapter via ContextVar]
        M6[Catch ToolError]
        M7[response.model_dump]
        M8[add_v2_compat]
        M1 --> M2 --> M3 --> M4 --> M5 --> M6 --> M7 --> M8
    end

    subgraph A2A["A2A Handler"]
        A1[Extract auth token]
        A2[Validate parameters manually]
        A3[Call _impl]
        A4[set_current_tenant]
        A5[get_current_tenant from ContextVar]
        A6[get_adapter via ContextVar]
        A7[Return error dict]
        A8[Serialize for A2A]
        A1 --> A2 --> A3 --> A4 --> A5 --> A6 --> A7 --> A8
    end

    style M1 fill:#f9c74f
    style A1 fill:#f9c74f
    style M3 fill:#f4845f
    style M4 fill:#f4845f
    style A4 fill:#f4845f
    style A5 fill:#f4845f
    style M6 fill:#c1121f
    style A7 fill:#c1121f
    style M7 fill:#90be6d
    style M8 fill:#90be6d
    style A8 fill:#90be6d
    style A2 fill:#577590

Loading

🟡 Auth 🔴 Tenant ContextVar 🔵 Validation 🟢 Serialization — all duplicated across transports and mixed into _impl()

After: each concern lives in one layer

flowchart TD
    subgraph boundary["Transport Boundary (MCP / A2A / REST)"]
        direction TB
        B1["Auth extraction → principal_id"]
        B2["Tenant resolution → LazyTenantContext"]
        B3["Build ResolvedIdentity"]
        B1 --> B2 --> B3
    end

    subgraph impl["Business Logic (_impl)"]
        direction TB
        I1["Receive (request, identity)"]
        I2["identity.tenant_id — no DB query"]
        I3["identity.tenant.ad_server — lazy DB load"]
        I4["get_adapter(tenant=tenant) — explicit"]
        I5["Raise AdCPError — transport-agnostic"]
        I6["Return typed response model"]
        I1 --> I2 --> I3 --> I4 --> I5 --> I6
    end

    subgraph translate["Error Boundary"]
        direction TB
        E1["AdCPError → ToolError (MCP)"]
        E2["AdCPError → ServerError (A2A)"]
        E3["AdCPError → JSON 4xx (REST)"]
    end

    subgraph serial["Response"]
        direction TB
        S1["Framework serializes model"]
        S2["Version compat transform"]
    end

    boundary --> impl --> translate --> serial

    style B1 fill:#f9c74f
    style B2 fill:#f9c74f
    style B3 fill:#f9c74f
    style I1 fill:#4895ef
    style I2 fill:#4895ef
    style I3 fill:#4895ef
    style I4 fill:#4895ef
    style I5 fill:#4895ef
    style I6 fill:#4895ef
    style E1 fill:#c1121f,color:#fff
    style E2 fill:#c1121f,color:#fff
    style E3 fill:#c1121f,color:#fff
    style S1 fill:#90be6d
    style S2 fill:#90be6d

Loading

🟡 Auth + tenant — resolved once at boundary 🔵 Business logic — pure functions, no global state 🔴 Errors — translated at the edge 🟢 Serialization — framework handles it

Data flow through the layers

sequenceDiagram
    participant T as Transport<br/>(MCP / A2A / REST)
    participant B as Boundary<br/>(resolve_identity)
    participant I as _impl()<br/>(business logic)
    participant L as LazyTenantContext
    participant DB as Database

    T->>B: incoming request + auth token
    B->>B: extract principal_id from token
    B->>B: resolve tenant_id from header/subdomain
    B->>B: create LazyTenantContext(tenant_id)
    Note over B: No DB query yet
    B->>I: _impl(request, ResolvedIdentity)

    I->>L: identity.tenant.tenant_id
    L-->>I: "test_tenant" (immediate, no DB)

    I->>L: identity.tenant.ad_server
    L->>DB: get_tenant_by_id("test_tenant")
    DB-->>L: {tenant_id, name, ad_server, ...}
    Note over L: Cached — subsequent<br/>accesses don't hit DB
    L-->>I: "gam"

    I->>I: get_adapter(tenant=tenant)
    I->>I: business logic...
    I-->>T: GetProductsResponse (typed model)

    alt Success
        T->>T: framework serializes response
    else AdCPError raised
        T->>T: translate to transport-specific error
    end

Loading

Lazy tenant loading detail

~85% of call paths only need tenant_id, which is available without a DB query. Full tenant data loads transparently on first access to any other field.

stateDiagram-v2
    [*] --> Unloaded: LazyTenantContext(tenant_id)

    state Unloaded {
        [*] --> Available
        Available: tenant_id ✅ (immediate)
        Available: __contains__ ✅ (no DB)
        Available: __bool__ ✅ (always True)
    }

    Unloaded --> Loaded: Access .name, .ad_server,<br/>.approval_mode, etc.

    state Loaded {
        [*] --> Resolved
        Resolved: All 22 fields available
        Resolved: Backed by TenantContext model
        Resolved: set_current_tenant() called once
        Resolved: Cached — no repeat queries
    }

    note right of Unloaded: ~85% of _impl calls<br/>stay here
    note right of Loaded: Only when _impl needs<br/>full tenant config

Loading

Concrete changes

1. Unified process — Single FastAPI app replaces 3 subprocesses

• FastMCP v2 → v3.0.2, mounted at /mcp via http_app()
• A2A mounted via A2AFastAPIApplication.add_routes_to_app()
• Flask Admin mounted via WSGIMiddleware
• nginx configs → single upstream, Docker Compose → single service

2. Identity resolution — Auth + tenant resolved once at boundary

• ResolvedIdentity — frozen Pydantic model carrying resolved auth + tenant context
• TenantContext / LazyTenantContext — typed model replacing raw dicts, defers DB load until needed (~85% of calls never hit DB)
• get_adapter() takes explicit tenant parameter (no ContextVar reads)
• All 12 _impl modules use identity.tenant exclusively
• Single pure ASGI UnifiedAuthMiddleware replaces 3 separate auth middlewares
• Each transport delegates to resolve_identity() via its own mechanism: FastMCP Middleware (MCP), _resolve_a2a_identity (A2A), FastAPI Depends (REST)

3. Error boundary — Transport-agnostic errors in business logic

• AdCPError hierarchy (Validation, Authentication, Authorization, Adapter) replaces ToolError in 42+ sites across _impl functions
• Error boundary middleware translates at each transport edge
• _impl never sees ToolError, ServerError, or HTTP status codes

4. Structural enforcement — 77 tests verify architecture invariants

• _impl functions never touch ContextVars
• resolve_identity() internals are pure (no set_current_tenant() side effects)
• Token extraction is consistent across all transports
• Auth infrastructure uses shared constants and immutable types

5. REST API — Full feature parity, for free

Because all business logic lives in shared _impl functions, adding a REST transport was a single 183-line file (src/routes/api_v1.py) with zero changes to business logic. It has complete feature parity with MCP and A2A — same 10 tools, same auth, same validation, same errors — just JSON-RPC over HTTP instead of MCP/A2A protocols. Swagger UI auto-generated at /docs, OpenAPI spec at /openapi.json. This layer is effectively maintenance-free: when an _impl changes, all three transports pick it up automatically.

Benefits

• Simplicity: Adding a new transport (REST took 1 file) doesn't require reimplementing auth/tenant/errors
• Security: Auth is resolved once at the boundary, not scattered across handlers. Debug endpoints gated behind ADCP_TESTING
• Stability: No ContextVar race conditions, no global state mutations in business logic, lazy tenant loading prevents unnecessary DB queries

Test suite health

The test suite was in rough shape — 138 warnings, flaky failures from resource leaks, unawaited coroutines, stale mocks, and tests that didn't actually test what they claimed. This PR fixes all of that:

Metric	Before	After
Passing tests	~2,474	3,490
Warnings	138	0
Failures/flaky	Several	0
Unit suite time	54s	17s
Full suite parallelism	Sequential	5 suites in parallel (tox)
Coverage reporting	None	51% line + branch

What was fixed

• Resource leaks: Unclosed async generators, HTTPServer sockets not released, asyncio ResourceWarnings from dependent libraries
• Stale mocks: Tests patching deleted functions, xfail markers on passing tests, MagicMock objects where ResolvedIdentity was needed
• Missing coverage: 29 new unit tests for update_performance_index and get_adcp_capabilities, 8 new e2e tests for discovery endpoints and multi-tenant isolation
• Broken context propagation: list_authorized_properties MCP wrapper silently dropped the context parameter
• Slow tests: time.sleep in retry/backoff tests replaced with mocks (54s → 17s)

Test infrastructure

• tox + tox-uv for parallel test orchestration with combined coverage
• ./run_all_tests.sh — one command: starts Docker, runs tox -p (5 suites in parallel), combines coverage, tears down Docker
• make test-stack-up / make test-stack-down — Docker lifecycle separated for fast iteration
• Coverage: htmlcov/index.html (HTML) + coverage.json (machine-readable), per-env isolation prevents SQLite race conditions
• Auto-pruning: test-results/ keeps only last 10 runs
• JSON reports: Per-suite test results in test-results/<timestamp>/

Test plan

• 3,490 tests pass (2,688 unit + 543 integration + 174 integration_v2 + 81 e2e + 4 ui)
• 0 warnings, 0 failures
• 51% line coverage (branch coverage enabled)
• make quality passes (ruff format + lint + mypy + unit tests)
• tox -e unit verified independently
• Full ./run_all_tests.sh run verified end-to-end
• 77 structural tests enforce architecture invariants (no ContextVar in _impl, pure resolver, consistent auth)
• docker compose up — unified stack serves all routes
• Admin UI accessible at /admin/
• Swagger UI at /docs lists all 10 REST endpoints

Nginx configuration simplification

Since all three services (MCP, A2A, Admin) now run in a single FastAPI process, the nginx configs were simplified from ~40 location blocks to catch-all patterns. Each config now uses a single location / with WebSocket-safe headers (via map $http_upgrade $connection_upgrade) instead of repeating identical proxy blocks per path. Total reduction: 957 → 472 lines (50%) across the three configs (development, single-tenant, multi-tenant).

Post-review hardening (29 commits)

The initial review surfaced gaps between the claimed architecture and the actual implementation — auth was described as "resolved once at the boundary" but each transport still had its own extraction logic, resolver internals still mutated ContextVars, and A2A had duplicated tenant detection with the wrong strategy order. 29 follow-up commits close these gaps.

Auth unification completed

• Replaced 3 separate auth middlewares with a single pure ASGI UnifiedAuthMiddleware
• MCP: auth centralized via FastMCP Middleware → resolve_identity()
• A2A: identity resolved once at transport boundary and threaded through all handlers; inline tenant detection replaced by delegation to resolve_identity()
• REST: converted to FastAPI Depends with Annotated types
• Eliminated ContextVar dual-write from A2A auth path

ContextVar side effects eliminated from resolver internals

• _detect_tenant() made pure — 6 set_current_tenant() calls removed
• get_principal_from_token() returns (principal_id, tenant_dict) tuple instead of mutating ContextVar
• resolve_identity() no longer falls back to get_current_tenant()
• set_current_tenant() now called only at transport boundaries, never inside the resolver

A2A handler correctness

• Fixed duplicate tenant detection that used the wrong strategy order
• Eliminated double header parsing in A2A middleware
• CallContextBuilder wired to thread SDK ServerCallContext to handler helpers
• Task IDs use UUID; removed debug log statements; stub methods raise ServerError

Infrastructure hardening

• AuthContext.headers wrapped in MappingProxyType (immutable after creation)
• AUTH_CONTEXT_STATE_KEY shared constant replaces string literal duplication
• Token extraction priority reconciled between middleware and resolve_identity()
• Test paths ported to pathlib for cross-platform portability
• Stale comments and docs updated to reflect post-migration architecture

🤖 Generated with Claude Code

[KonstantinMirin]

Fixed the time-based test (1st of March is in the past now), updated the test for authenticated context as it works differently now.
Also, marked adcp schema compliance mistakes as xpass because
(a) there is an issue with how we check it (against the latest schema, not client-declared schema version)
(b) we have a pending PR for upgrading to 3.6
(c) we really need to have a decision and a policy about the protocol versioning and releases.

[ChrisHuie]

would add the _is_valid_hostname(host) similar to above with apx_incoming_host

[ChrisHuie]

this imports the old get_product_catalog before your changes. Should be => from src.core.tools.products import get_product_catalog

[ChrisHuie]

webhook_verification and webhook_authenticator both use hmac for constant-time comparison over ==

would think about going something like this instead maybe...

import hmac

if tenant_obj and tenant_obj.admin_token and hmac.compare_digest(tenant_obj.admin_token, token):

[ChrisHuie]

Got this from claude but not sure if it 100% true. Would ask your agent to maybe investigate further to see if it is an issue?

TENANT-1 — TenantContext vs dict Type Mismatch (Medium)

The chain of events:

Step 1 — resolve_identity() wraps tenant dicts in TenantContext (src/core/resolved_identity.py:184-192):
tenant_model = tenant_context
if isinstance(tenant_context, dict) and "tenant_id" in tenant_context:
tenant_model = TenantContext.from_dict(tenant_context) # Now a Pydantic BaseModel, NOT a dict
return ResolvedIdentity(..., tenant=tenant_model, ...)

Step 2 — Transport boundaries pass it to set_current_tenant() without conversion:

src/core/auth_context.py:91-94 (REST):
if identity.tenant:
set_current_tenant(identity.tenant) # ← passes TenantContext (Pydantic model)

src/a2a_server/adcp_a2a_server.py:207-210 (A2A):
if identity.tenant:
set_current_tenant(identity.tenant) # ← same issue

Step 3 — But set_current_tenant is typed for dict (src/core/config_loader.py:184):
current_tenant: ContextVar[dict[str, Any] | None] = ContextVar(...) # line 57
def set_current_tenant(tenant_dict: dict[str, Any]): # line 184
current_tenant.set(tenant_dict)

Step 4 — Downstream code checks isinstance(tenant, dict) and gets False:

src/core/helpers/context_helpers.py:79 (ensure_tenant_context):
if isinstance(tenant, dict) and "tenant_id" in tenant: # ← FALSE for TenantContext!
return tenant

Falls through to Step 4: redundant DB lookup by tenant_id

Why it doesn't crash: TenantContext implements getitem, get(), contains, keys() (lines 64-82 of tenant_context.py), so tenant["tenant_id"] works. Most code paths survive via fallbacks. But
ensure_tenant_context() does a wasted DB query on every REST/A2A request because it doesn't recognize TenantContext as valid tenant data.

Fix: Either update isinstance checks to isinstance(tenant, (dict, TenantContext)), or convert to dict before calling set_current_tenant():
set_current_tenant(identity.tenant.model_dump() if hasattr(identity.tenant, 'model_dump') else identity.tenant)

[ChrisHuie]

why not identity=identity ?

[KonstantinMirin]

Good catch — fixed in ed27fc9. Now passes identity=identity directly (matching update_media_buy at line 1825) and removes the unnecessary tool_context construction.

[KonstantinMirin]

@ChrisHuie than you for catching a few important issues. I went on looking for similar and found some. Filed a ticket for everything not addressed yet here #1078

I would like to get to it after merging this and outstanding 3.6 migration work. Reason being simple - fixing now would force merges/reviews in all the derived PRs. Even though it is a good thing to do, it doesn't really matter if we address this now or then.

That said, if the decision is to address it now, I'll get to it.

[ChrisHuie]

do we want these to raise ToolError or AdCPError

[KonstantinMirin]

Sharp catch — you're exactly right. ToolError is an MCP transport concern (fastmcp.exceptions) and should never appear in shared _impl() functions that are called by MCP, A2A, and REST transports.

Fixed in 48d5bf3:

• Replaced raise ToolError("Identity is required") → raise AdCPAuthenticationError(...)
• Replaced raise ToolError("account_id filtering is not yet supported") → raise AdCPValidationError(...)
• Added regression tests that guard against MCP types leaking back into _impl code
• Also cleaned up stale ToolError references in docstrings/comments across products.py, creative_helpers.py, and media_buy_create.py (code already used domain exceptions, comments were just outdated)

The MCP wrapper at line 242 still raises ToolError correctly — that IS the transport boundary where with_error_logging translates domain exceptions → ToolError for the MCP wire format.