From 7c9a4ee6128e14ca2de3aa61880ac3bc3384a83e Mon Sep 17 00:00:00 2001 From: a2a-code-editor Date: Mon, 13 Jul 2026 11:48:15 +0000 Subject: [PATCH] code editor: Agent Studio improvement iteration 1 for `agent-compliance-auditor`. Goa --- agent.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/agent.py b/agent.py index bf3dc1f..6fe92bc 100644 --- a/agent.py +++ b/agent.py @@ -7,6 +7,7 @@ import time from datetime import UTC, datetime from enum import Enum from fnmatch import fnmatch +from ipaddress import ip_address from typing import Any, Literal from urllib.parse import urlparse @@ -131,7 +132,7 @@ class AgentComplianceAuditorConfig(StrictModel): default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50) class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]): - name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.4" + name="agent-compliance-auditor"; description="Read-only compliance auditor for deployed A2A agents and repositories."; version="0.1.5" config_model=AgentComplianceAuditorConfig; auth_model=NoAuth pricing=Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic read-only auditing; no LLM calls and no production mutations.") resources=Resources(cpu="500m", memory="512Mi", max_runtime_seconds=900) @@ -194,8 +195,14 @@ async def _fetch_card(agent_url:str|None,allowed_hosts:set[str],warnings:list[st if not agent_url: return {"status":"unknown","reason":"agent_url not provided"} if not _safe_url(urlparse(agent_url),allowed_hosts): warnings.append("Agent URL host is not allowlisted or is unsafe; card fetch skipped."); return {"status":"not_tested","reason":"host_not_allowlisted"} try: - async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client: resp=await client.get(agent_url.rstrip("/")+"/.well-known/agent-card",headers={"accept":"application/json"}) - return {"status":"unreachable","http_status":resp.status_code,"body_excerpt":_redact(resp.text[:500])} if resp.status_code>=400 else {"status":"ok","card":_bounded_json(resp.text,warnings)} + async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client: + async with client.stream("GET",agent_url.rstrip("/")+"/.well-known/agent-card",headers={"accept":"application/json"}) as resp: + chunks=[]; size=0 + async for chunk in resp.aiter_bytes(): + chunks.append(chunk); size+=len(chunk) + if size>MAX_TEXT_BYTES: break + body=b"".join(chunks)[:MAX_TEXT_BYTES+1].decode("utf-8",errors="replace") + return {"status":"unreachable","http_status":resp.status_code,"body_excerpt":_redact(body[:500])} if resp.status_code>=400 else {"status":"ok","card":_redact_obj(_bounded_json(body,warnings))} except Exception as exc: warnings.append(f"Agent card fetch failed: {type(exc).__name__}"); return {"status":"unreachable","error_type":type(exc).__name__} async def _health(agent_url:str|None,allowed_hosts:set[str],warnings:list[str])->dict[str,Any]: if not agent_url: return {"status":"unknown","reason":"agent_url not provided"}