bobseares-2/agent-compliance-auditor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

code editor: Agent Studio improvement iteration 1 for agent-compliance-auditor. Goa

Browse Source
This commit is contained in:
a2a-code-editor
2026-07-13 11:48:15 +00:00
parent fcc5afbd0a
commit 7c9a4ee612
1 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
agent.py
View File

@@ -7,6 +7,7 @@ import time
from datetime import UTC, datetime from datetime import UTC, datetime
from enum import Enum from enum import Enum
from fnmatch import fnmatch from fnmatch import fnmatch
from ipaddress import ip_address
from typing import Any, Literal from typing import Any, Literal
from urllib.parse import urlparse from urllib.parse import urlparse
@@ -131,7 +132,7 @@ class AgentComplianceAuditorConfig(StrictModel):
default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50) default_allowed_namespaces: list[str] = Field(default_factory=list, max_length=50)
class AgentComplianceAuditor(A2AAgent[AgentComplianceAuditorConfig, NoAuth]): 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 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.") 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) 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 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"} 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: 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"}) async with httpx.AsyncClient(timeout=httpx.Timeout(HTTP_TIMEOUT_SECONDS),follow_redirects=False) as client:
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 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__} 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]: 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"} if not agent_url: return {"status":"unknown","reason":"agent_url not provided"}