platform/agent-harness-spec
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial harness spec v1.0.0 + reference template

Browse Source 
Adds the normative contract every platform agent must conform to, plus a
working reference template under template/ that ticks every box out of
the box.

Spec is derived from the production agents shipped in
cjot-backstage-az/agents/ on branch sandbox/jonathan:
  - agent-registry, agent-factory, decomposer, discovery-agent,
    golden-path, modernization-factory, modernization-factory-v2,
    policy-transformer, scaffold-agent, support-intake-agent, the-watcher

Contents:
  - SPEC.md          normative contract (13 sections + conformance checklist)
  - CHANGELOG.md     spec versioning (1.0.0)
  - docs/
      registration.md   self-registration with the agent-gateway
      observability.md  OTel logs/metrics/traces wiring
      manifest.md       /.well-known/agent.json schema + AgentSkill
                        serialisation (pydantic vs protobuf)
      kubernetes.md     k8s deployment shape, mandatory cross-refs,
                        per-namespace agents, resource sizing
      deployment.md     .image-version, ACR build, deploy scripts, rollback
      deviations.md     tracked debts against the spec
  - template/
      .env.local.example, .gitignore, .image-version (0.1.0),
      Dockerfile (python:3.12-slim, non-root UID 1001, HEALTHCHECK),
      requirements.txt (harness floor + optional LLM stack),
      app/agent.py (Starlette entry, lifespan + self-registration,
                    defensive _skill_dict for pydantic vs protobuf),
      app/logging_setup.py (canonical OTel log bridge — copy verbatim),
      app/metrics.py (meter + example counter/histogram),
      app/skills.py (AGENT_CONFIG + AgentSkill list),
      k8s/configmap.yaml + deployment.yaml (Deployment + Service,
                                            OTel annotations + 6-var env block,
                                            agents-sa + agents-kv-spc bindings),
      scripts/deploy.sh (auto-bump + az acr build + apply + rollout),
      scripts/full-deploy.sh (preflight + deploy + post-deploy smoke),
      scripts/deploy-local.sh (docker/podman + .env.local)
This commit is contained in:
Jonathan Boniface
2026-06-09 15:54:18 +01:00
parent c87dd3fa55
commit e6f10ba8c9
25 changed files with 1933 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
template/app/__init__.py Normal file
View File

185
template/app/agent.py Normal file
View File

@@ -0,0 +1,185 @@
"""<AgentName> — Starlette entry point.
This module is the canonical agent harness. It implements the cross-cutting
plumbing every agent on the platform must ship:
- GET /health liveness/readiness probe
- GET /.well-known/agent.json A2A discovery manifest
- lifespan-driven self-registration with the agent-gateway
- OTel log bridge wired at import time
- LangChain instrumentation wired before any LLM client construction
Replace the example `/echo` route with your domain endpoints.
"""
from __future__ import annotations
import asyncio
import logging
import os
import time
from contextlib import asynccontextmanager
import httpx
from dotenv import load_dotenv
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
try:
from openinference.instrumentation.langchain import LangChainInstrumentor
except ImportError:
LangChainInstrumentor = None # type: ignore[assignment]
from .logging_setup import configure_otlp_log_handler
from .metrics import echo_duration, echoes
from .skills import AGENT_CONFIG, AGENT_SKILLS
# ─── One-shot startup wiring ────────────────────────────────────────────
load_dotenv(".env.local")
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
configure_otlp_log_handler(getattr(logging, LOG_LEVEL.upper(), logging.INFO))
logger = logging.getLogger(__name__)
if LangChainInstrumentor is not None:
try:
LangChainInstrumentor().instrument()
except Exception as exc: # pragma: no cover — instrumentation failure is non-fatal
logger.warning("LangChainInstrumentor failed to attach: %s", exc)
# ─── Configuration ──────────────────────────────────────────────────────
REGISTRY_URL = os.getenv("REGISTRY_URL", "").strip()
AGENT_SELF_URL = os.getenv(
"AGENT_SELF_URL",
"http://<agent-name>.agents.svc.cluster.local",
).strip()
PORT = int(os.getenv("PORT", "8000"))
# ─── Skill serialisation (handles pydantic + protobuf AgentSkill) ───────
def _skill_dict(s) -> dict:
"""Serialise an A2A skill to a JSON-safe dict.
a2a-sdk exposes AgentSkill as a protobuf Message in current releases, so
`s.model_dump()` raises AttributeError. Handle pydantic v2, pydantic v1,
protobuf, and fall back to attribute scraping.
"""
if hasattr(s, "model_dump"):
return s.model_dump()
if hasattr(s, "DESCRIPTOR"):
try:
from google.protobuf.json_format import MessageToDict
return MessageToDict(s, preserving_proto_field_name=True)
except Exception: # pragma: no cover
pass
if hasattr(s, "dict") and callable(getattr(s, "dict")):
try:
return s.dict()
except Exception: # pragma: no cover
pass
return {
"id": getattr(s, "id", None),
"name": getattr(s, "name", None),
"description": getattr(s, "description", ""),
"tags": list(getattr(s, "tags", []) or []),
"examples": list(getattr(s, "examples", []) or []),
}
# ─── Route handlers ─────────────────────────────────────────────────────
async def health_check(request: Request) -> JSONResponse:
return JSONResponse({
"status": "healthy",
"service": AGENT_CONFIG["name"],
"version": AGENT_CONFIG["version"],
})
async def agent_manifest(request: Request) -> JSONResponse:
"""GET /.well-known/agent.json — discovery manifest consumed by agent-gateway."""
return JSONResponse({
"name": AGENT_CONFIG["name"],
"version": AGENT_CONFIG["version"],
"description": AGENT_CONFIG["description"],
"url": "/",
"skills": [_skill_dict(s) for s in AGENT_SKILLS],
"capabilities": AGENT_CONFIG["capabilities"],
})
async def echo(request: Request) -> JSONResponse:
"""POST /echo — example endpoint. Replace with your domain logic."""
started = time.perf_counter()
try:
body = await request.json()
except Exception:
echoes.add(1, {"outcome": "bad_request"})
return JSONResponse({"error": "invalid JSON body"}, status_code=400)
try:
response = {"received": body}
echoes.add(1, {"outcome": "success"})
return JSONResponse(response)
except Exception as exc:
echoes.add(1, {"outcome": "error"})
logger.exception("echo failed: %s", exc)
return JSONResponse({"error": str(exc)}, status_code=500)
finally:
echo_duration.record(time.perf_counter() - started)
# ─── Self-registration with the agent-gateway ───────────────────────────
async def _register_with_registry() -> None:
"""POST our endpoint to the registry; it will GET /.well-known/agent.json back."""
if not REGISTRY_URL:
logger.info("REGISTRY_URL not set — skipping self-registration")
return
await asyncio.sleep(2) # let uvicorn finish binding before the gateway calls back
url = f"{REGISTRY_URL.rstrip('/')}/agents/register-url"
payload = {"endpoint": AGENT_SELF_URL}
for attempt in range(3):
try:
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(url, json=payload)
if resp.status_code in (200, 201):
logger.info("Self-registered with agent-registry at %s", REGISTRY_URL)
return
logger.warning(
"Registry registration attempt %d: HTTP %d — %s",
attempt + 1, resp.status_code, resp.text[:200],
)
except Exception as exc:
logger.warning("Registry registration attempt %d failed: %s", attempt + 1, exc)
await asyncio.sleep(5)
logger.error("Failed to self-register with agent-registry after 3 attempts")
@asynccontextmanager
async def lifespan(app: Starlette):
register_task = asyncio.create_task(_register_with_registry())
try:
yield
finally:
register_task.cancel()
# ─── Application ────────────────────────────────────────────────────────
app = Starlette(
routes=[
Route("/health", health_check, methods=["GET"]),
Route("/.well-known/agent.json", agent_manifest, methods=["GET"]),
Route("/echo", echo, methods=["POST"]),
],
lifespan=lifespan,
)
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.agent:app", host="0.0.0.0", port=PORT, log_level=LOG_LEVEL.lower())

34
template/app/logging_setup.py Normal file
View File

@@ -0,0 +1,34 @@
"""Bridge Python stdlib logging into the OTel LoggerProvider.
The OTel Operator's Python auto-instrumentation creates a LoggerProvider with
an OTLP BatchLogRecordProcessor, but it does not attach a LoggingHandler bridge
to Python's root logger — so stdlib logger.info(...) calls never reach the
collector. Call configure_otlp_log_handler() once at startup to close that gap.
This module is part of the platform agent harness — do not edit. The canonical
copy lives in the agent-harness-spec repo and is copied verbatim into each
agent. If you need to change the bridge, change it there and re-roll.
"""
from __future__ import annotations
import logging
def configure_otlp_log_handler(level: int = logging.INFO) -> None:
try:
from opentelemetry._logs import get_logger_provider
from opentelemetry.sdk._logs import LoggingHandler
except ImportError:
return
provider = get_logger_provider()
if provider is None or type(provider).__name__ == "NoOpLoggerProvider":
return
root = logging.getLogger()
if any(isinstance(h, LoggingHandler) for h in root.handlers):
return
root.addHandler(LoggingHandler(level=level, logger_provider=provider))
if root.level == logging.NOTSET or root.level > level:
root.setLevel(level)

33
template/app/metrics.py Normal file
View File

@@ -0,0 +1,33 @@
"""OTel metric instruments for <agent-name>.
All metrics flow through the global MeterProvider installed by the OTel
Operator's init container. The provider exports OTLP to the cluster tier-1
collector, which forwards metrics to Prometheus via tier-2's remote-write.
Naming convention (see SPEC §7.4):
agent.<agent_type>.<noun> — counters (unit="1")
agent.<agent_type>.<noun>.duration — histograms in seconds (unit="s")
Replace `<agent_name>` and the example instruments with your own.
"""
from __future__ import annotations
from opentelemetry import metrics
# The (name, version) pair becomes `otel_scope_name` and `otel_scope_version`
# attributes on every metric — useful for filtering in PromQL.
_meter = metrics.get_meter("<agent-name>", "0.1.0")
# ─── Example: an "echo" operation counter and latency histogram ─────────
echoes = _meter.create_counter(
"agent.<agent_name>.echoes",
unit="1",
description="Echo requests by outcome.",
)
echo_duration = _meter.create_histogram(
"agent.<agent_name>.echo.duration",
unit="s",
description="End-to-end /echo latency.",
)

40
template/app/skills.py Normal file
View File

@@ -0,0 +1,40 @@
"""A2A skill definitions + agent config for <agent-name>.
See docs/manifest.md for the manifest schema this feeds into.
"""
from __future__ import annotations
from a2a.types import AgentSkill
# ─── Agent identity ─────────────────────────────────────────────────────
# The version here MUST match the contents of ../.image-version and the
# `service.version` token in OTEL_RESOURCE_ATTRIBUTES.
AGENT_CONFIG: dict = {
"name": "<AgentName>",
"version": "0.1.0",
"description": (
"One- to three-sentence description of what this agent does, "
"phrased for a non-engineering audience."
),
"capabilities": {
"streaming": False,
"conversational": False,
"direct_api": True,
},
}
# ─── Skills ─────────────────────────────────────────────────────────────
AGENT_SKILLS: list[AgentSkill] = [
AgentSkill(
id="echo",
name="Echo",
description=(
"A trivial skill that returns the request payload verbatim. "
"Replace with your real skills."
),
tags=["example", "echo"],
examples=[
"Echo: hello world",
],
),
]