12c6347643
All checks were successful
build / build (push) Successful in 5s
Two skills:
- echo(text): trivial sub-second response. Used by approval-mode tests
that need a fast handoff target.
- try_scope_expansion(reason, read_patterns, ttl_seconds, mode):
deliberately calls ctx.request_scope() with caller-supplied args so
e2e tests can exercise the orchestrator's scope-negotiation flow
end-to-end without inventing a real workload that happens to need
extra files.
Standard A2A scaffold: Dockerfile FROM a2a-pack-base, /invoke/{skill} +
/.well-known/agent-card served by the SDK's serve/asgi.py adapter, deploy/
manifests for k8s, Gitea Actions workflow that builds + bumps the image
SHA.
Public, low resources (50m CPU / 128Mi mem). Not exposed via the public
ingress in any meaningful way — only the in-cluster Service is consulted
by the orchestrator's call_agent.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""test-helper agent — exercises platform features for end-to-end tests.
|
|
|
|
Not a user-facing agent. Lives on the platform so e2e tests can target
|
|
predictable, fast behaviour without bringing the chart pipeline along
|
|
for every assertion.
|
|
|
|
Skills:
|
|
- ``echo``: trivial fast response. For approval-mode tests that just
|
|
need *some* handoff to pause on.
|
|
- ``try_scope_expansion``: deliberately calls ``ctx.request_scope()``
|
|
so the orchestrator's scope-negotiation path can be exercised.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from a2a_pack import A2AAgent, NoAuth, RunContext, skill
|
|
from a2a_pack.context import ScopeDenied
|
|
|
|
|
|
class TestHelperConfig(BaseModel):
|
|
pass
|
|
|
|
|
|
class TestHelperAgent(A2AAgent[TestHelperConfig, NoAuth]):
|
|
name = "test-helper"
|
|
description = (
|
|
"Platform feature exerciser for e2e tests: cheap echo + a deliberate "
|
|
"ctx.request_scope() caller."
|
|
)
|
|
version = "0.1.0"
|
|
|
|
config_model = TestHelperConfig
|
|
auth_model = NoAuth
|
|
tools_used = ("e2e",)
|
|
|
|
@skill(
|
|
description=(
|
|
"Echo the text back. Trivial, sub-second response — used by "
|
|
"tests that need to exercise the handoff/approval mechanics "
|
|
"without paying for chart rendering."
|
|
),
|
|
tags=["e2e", "echo"],
|
|
)
|
|
async def echo(self, ctx: RunContext[NoAuth], text: str = "ping") -> dict:
|
|
await ctx.emit_progress(f"echoing {len(text)} chars")
|
|
bucket = getattr(ctx.workspace, "bucket", None)
|
|
return {"ok": True, "echoed": text, "bucket": bucket}
|
|
|
|
@skill(
|
|
description=(
|
|
"Call ctx.request_scope() with the caller-supplied read patterns "
|
|
"+ ttl, then report the platform's response. Lets tests verify "
|
|
"the orchestrator's scope-negotiation flow end-to-end without "
|
|
"needing a real workload that happens to need extra files."
|
|
),
|
|
tags=["e2e", "scope-request"],
|
|
allow_scope_expansion=True,
|
|
)
|
|
async def try_scope_expansion(
|
|
self,
|
|
ctx: RunContext[NoAuth],
|
|
reason: str = "e2e test of scope negotiation",
|
|
read_patterns: list[str] = ["reference/**"],
|
|
ttl_seconds: int = 60,
|
|
mode: str = "read_only",
|
|
) -> dict:
|
|
original = list(getattr(ctx.workspace, "allow_patterns", ()))
|
|
await ctx.emit_progress(
|
|
f"requesting scope expansion: read={read_patterns} ttl={ttl_seconds}s"
|
|
)
|
|
try:
|
|
grant = await ctx.request_scope(
|
|
reason=reason,
|
|
read=read_patterns,
|
|
ttl_seconds=ttl_seconds,
|
|
mode=mode,
|
|
)
|
|
except ScopeDenied as exc:
|
|
await ctx.emit_progress(f"scope denied: {exc}")
|
|
return {
|
|
"ok": False,
|
|
"denied": True,
|
|
"reason": str(exc),
|
|
"original_allow": original,
|
|
}
|
|
|
|
new_allow = list(getattr(ctx.workspace, "allow_patterns", ()))
|
|
await ctx.emit_progress(f"scope granted: now {new_allow}")
|
|
return {
|
|
"ok": True,
|
|
"original_allow": original,
|
|
"new_allow": new_allow,
|
|
"new_grant_id": grant.grant_id,
|
|
"new_mode": grant.mode if isinstance(grant.mode, str)
|
|
else grant.mode.value,
|
|
"new_ttl_seconds": grant.expires_at - grant.issued_at,
|
|
}
|