bobseares-2/customer-integration-engineer
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 03:33:02 +00:00
parent 87b7fa979c
commit 18f57818d4
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
agent.py
View File

@@ -573,14 +573,19 @@ class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, No
network_calls = 0
warnings: list[str] = []
for test in request.endpoint_tests:
validation = await _validate_endpoint_target(test.url, request.external_host_allowlist)
idem = test.idempotency_key or _plan_digest({"integration_id": request.integration_id, "test": test.model_dump(mode="json")})[:32]
if dry_run or not request.allow_network:
validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist)
if validation:
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem))
warnings.append(validation)
else:
step_results.append(AcceptanceStepResult(name=test.name, status="skipped", attempts=0, detail="dry_run/default safe mode: endpoint metadata was validated but no DNS lookup or external request was sent", idempotency_key=idem))
continue
if dry_run or not request.allow_network:
step_results.append(AcceptanceStepResult(name=test.name, status="skipped", attempts=0, detail="dry_run/default safe mode: external request was validated but not sent", idempotency_key=idem))
validation = await _validate_endpoint_target(test.url, request.external_host_allowlist)
if validation:
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem))
warnings.append(validation)
continue
if live_network_block:
step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=live_network_block, idempotency_key=idem))