Agents & skills
An a2a agent is one Python class that subclasses
A2AAgent and exposes one or more @skill-decorated
methods.
Required class attributes
class MyAgent(A2AAgent[MyConfig, MyAuth]):
name = "my-agent" # globally unique on the platform
description = "..."
version = "0.1.0" # semver, surfaces on Card + dashboard
config_model = MyConfig # Pydantic; passed as `self.config`
auth_model = MyAuth # Pydantic; bound to `ctx.auth` per callSkills
Decorate async methods. The first parameter (after self) must be a
RunContext. The rest must be type-annotated;
their types become the skill's input JSON Schema.
@skill(
description="Generate a chart.",
tags=["visualization"],
timeout_seconds=120,
allow_scope_expansion=False,
)
async def chart(
self,
ctx: RunContext[NoAuth],
prompt: str,
data_path: str | None = None,
) -> dict:
await ctx.emit_progress(f"working on: {prompt}")
return {"ok": True, "path": "outputs/chart.png"}The decorator:
- captures the signature with
inspect - builds a
pydantic.TypeAdapterper parameter - emits a JSON Schema with
additionalProperties: falseandrequiredpopulated from non-default args - publishes the resulting
SkillCardon your agent'sAgentCard
Input validation, two levels
- Pydantic at the agent. Every
/invoke/{skill}call runs your args throughadapter.validate_python(...)before the handler fires. Bad types → HTTP 400 (SkillInputError). - Strict tool schema at the LLM. The orchestrator can register each skill as an OpenAI-strict function tool — the model gateway refuses to emit calls that don't conform.
Optional opt-ins
| Attribute / flag | What it enables |
|---|---|
@skill(allow_scope_expansion=True) |
Skill may call ctx.request_scope(...) mid-flight |
@skill(stream=True) |
Skill emits multiple events; the platform SSE-bridges them |
workspace_access = WorkspaceAccess.dynamic(...) |
The platform attaches a workspace client to ctx |
egress = EgressPolicy(allow_hosts=...) |
Limit outbound network from the sandbox |
tools_used = ("sandbox", "matplotlib") |
Surfaces on the Card; informational |
Lifecycle
A2AAgent.startup(ctx) and A2AAgent.shutdown(ctx) run once each per
process. Use for warm-up (load a model, hydrate a cache) or graceful
shutdown.