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

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-13 11:52:22 +00:00
parent e11edd9c50
commit 6129481e8c
1 changed files with 6 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
agent.py
View File

@@ -132,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.6" 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)
@@ -335,6 +335,11 @@ def _redact(text:str)->str:
out=text out=text
for pat in SECRET_PATTERNS: out=pat.sub(_redact_match,out) for pat in SECRET_PATTERNS: out=pat.sub(_redact_match,out)
return out return out
def _redact_obj(value:Any)->Any:
if isinstance(value,str): return _redact(value)
if isinstance(value,list): return [_redact_obj(v) for v in value]
if isinstance(value,dict): return {str(k): _redact_obj(v) for k,v in value.items()}
return value
def _sha(text:str)->str: return hashlib.sha256(text.encode("utf-8",errors="replace")).hexdigest() def _sha(text:str)->str: return hashlib.sha256(text.encode("utf-8",errors="replace")).hexdigest()
def _dump(value:Any)->Any: return value.model_dump(mode="json") if isinstance(value,BaseModel) else value def _dump(value:Any)->Any: return value.model_dump(mode="json") if isinstance(value,BaseModel) else value
async def _event(ctx:RunContext[NoAuth],kind:str,**payload:Any)->None: async def _event(ctx:RunContext[NoAuth],kind:str,**payload:Any)->None: