feat: an A2A facade for the n8n agents, and Approvisionnement's first workflow
The adapter is written once and parameterised, so both n8n agents get the same A2A surface and differ only in the workflow behind the webhook. It owns the protocol and nothing else: no Keycloak client, no credentials, and the end user's token forwarded in a header so a workflow that logs its input does not log the token with it. escalate: true fails the task rather than completing it, because an autonomous agent meeting a rule that needs approval must hand it back, not decide. Serving this from the gateway was the repo's stated preference and does not reach: A2A returns a Task with a lifecycle, and a proxy cannot synthesise that from a webhook's single reply. The check_availability workflow does the token exchange itself, using the agent's client secret from the pod environment so the committed JSON holds no secret, then talks MCP in two calls — initialize to obtain the session id from a response header, then tools/call. It reads the product code with a regex rather than a model: this agent's critical path is rules, and a question naming no product escalates instead of being guessed at. MCP rather than the REST route, which would have been one HTTP call instead of two: the per-tool ACL by consumer group lives on the MCP route only. Going through REST would give an autonomous agent broader access than a conversational one, which is exactly backwards. Status: the adapter is tested (13 tests) and the identity chain is verified by hand, but the workflow fails at its token exchange and is not yet working — see the note in workflows/. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Stéphane Tailland
@@ -0,0 +1,20 @@
|
||||
"""Entry point: `python -m a2a_n8n_adapter`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import uvicorn
|
||||
|
||||
from .config import Settings
|
||||
from .server import build_app
|
||||
|
||||
|
||||
def main() -> None:
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
||||
settings = Settings() # type: ignore[call-arg] # the agent's identity comes from the environment
|
||||
uvicorn.run(build_app(settings), host=settings.host, port=settings.port)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,57 @@
|
||||
"""One adapter, configured per agent.
|
||||
|
||||
Both n8n agents get the same image and differ only by environment. That is the
|
||||
point: the A2A surface is identical for every agent in the estate, and what
|
||||
distinguishes Approvisionnement from Recouvrement is the workflow behind the
|
||||
webhook, not the protocol in front of it.
|
||||
|
||||
Nothing here is a secret. The adapter forwards the end user's token and holds
|
||||
none of its own — the Keycloak client secret belongs to the workflow, stored as
|
||||
an n8n credential, so the component that can mint tokens is the one that uses
|
||||
them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
model_config = SettingsConfigDict(env_prefix="ADAPTER_", env_file=".env")
|
||||
|
||||
# --- who this instance is ------------------------------------------------
|
||||
agent_name: str = Field(..., description="e.g. approvisionnement")
|
||||
agent_description: str = Field(...)
|
||||
agent_version: str = "0.1.0"
|
||||
|
||||
# --- the one skill it exposes -------------------------------------------
|
||||
# One skill per adapter instance, deliberately. An agent that grows a second
|
||||
# skill is an agent whose card should say so explicitly, and reviewing that
|
||||
# is the moment to ask whether it has quietly become a data proxy.
|
||||
skill_id: str = Field(..., description="e.g. check_availability")
|
||||
skill_name: str = Field(...)
|
||||
skill_description: str = Field(...)
|
||||
skill_examples: str = ""
|
||||
"""Newline-separated. They go in the card, which is what a calling agent's
|
||||
model reads to decide whether this is the specialist to ask."""
|
||||
|
||||
# --- the workflow behind it ---------------------------------------------
|
||||
webhook_url: str = Field(..., description="the n8n Webhook node's production URL")
|
||||
# A workflow is not a chat: it runs rules, calls APIs, and may wait on a
|
||||
# slow backend. The default here is generous for that, not for a model.
|
||||
webhook_timeout: float = 120.0
|
||||
|
||||
# --- transport -----------------------------------------------------------
|
||||
ca_bundle: str | None = None
|
||||
host: str = "0.0.0.0"
|
||||
port: int = 8080
|
||||
public_url: str = "http://localhost:8080/"
|
||||
|
||||
@property
|
||||
def verify(self) -> str | bool:
|
||||
return self.ca_bundle if self.ca_bundle else True
|
||||
|
||||
@property
|
||||
def examples(self) -> list[str]:
|
||||
return [line.strip() for line in self.skill_examples.splitlines() if line.strip()]
|
||||
@@ -0,0 +1,219 @@
|
||||
"""The A2A facade: an agent card, a task lifecycle, and a webhook behind it.
|
||||
|
||||
The repo's own notes proposed serving this from the gateway instead — a static
|
||||
Agent Card and a route to the webhook, no new component. That does not reach.
|
||||
A2A is not request/response: a caller gets a `Task` with an id, a context, and a
|
||||
state that can be `input-required` or `failed`, and none of that can be
|
||||
synthesised from a webhook's single reply by a proxy. The gateway would have to
|
||||
template the whole lifecycle, at which point it is an adapter written in
|
||||
configuration.
|
||||
|
||||
So: an adapter, written once and parameterised, which is the alternative those
|
||||
same notes anticipated. It is deliberately thin — it owns the protocol and
|
||||
nothing else. Every decision, every token exchange and every call to a domain
|
||||
happens in the workflow.
|
||||
|
||||
The a2a-sdk facts below were established against the running library, not read:
|
||||
|
||||
- `a2a.types` are protobuf messages, not pydantic models.
|
||||
- Use `DefaultRequestHandlerV2`; `DefaultRequestHandler` speaks 0.3 and rejects
|
||||
1.0 traffic. `enable_v0_3_compat=True` keeps older callers working.
|
||||
- An executor must enqueue a `Task` before any status update when
|
||||
`context.current_task` is None, or the SDK answers "Agent should enqueue Task
|
||||
before TaskStatusUpdateEvent".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
import httpx
|
||||
from a2a.server.agent_execution import AgentExecutor, RequestContext
|
||||
from a2a.server.events import EventQueue
|
||||
from a2a.server.request_handlers import DefaultRequestHandlerV2
|
||||
from a2a.server.routes.agent_card_routes import create_agent_card_routes
|
||||
from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
|
||||
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
|
||||
from a2a.types import (
|
||||
AgentCapabilities,
|
||||
AgentCard,
|
||||
AgentInterface,
|
||||
AgentSkill,
|
||||
Part,
|
||||
Task,
|
||||
TaskState,
|
||||
TaskStatus,
|
||||
)
|
||||
from starlette.applications import Starlette
|
||||
|
||||
from .config import Settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# The bearer the caller presents. It is the END USER's token, not the calling
|
||||
# agent's: the scoping model rests on `sub` staying the human throughout, while
|
||||
# `azp` names whichever agent is doing the asking.
|
||||
SUBJECT_TOKEN_HEADER = "authorization"
|
||||
|
||||
# What the workflow reads it from. A header rather than the body, so a workflow
|
||||
# that logs its input does not log the token with it.
|
||||
WORKFLOW_TOKEN_HEADER = "x-subject-token"
|
||||
|
||||
|
||||
def build_agent_card(settings: Settings) -> AgentCard:
|
||||
return AgentCard(
|
||||
name=settings.agent_name,
|
||||
description=settings.agent_description,
|
||||
version=settings.agent_version,
|
||||
default_input_modes=["text/plain"],
|
||||
default_output_modes=["text/plain"],
|
||||
capabilities=AgentCapabilities(streaming=False),
|
||||
supported_interfaces=[
|
||||
AgentInterface(
|
||||
url=settings.public_url,
|
||||
protocol_binding="JSONRPC",
|
||||
protocol_version="1.0",
|
||||
)
|
||||
],
|
||||
skills=[
|
||||
AgentSkill(
|
||||
id=settings.skill_id,
|
||||
name=settings.skill_name,
|
||||
description=settings.skill_description,
|
||||
tags=[settings.agent_name],
|
||||
examples=settings.examples,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
class WorkflowExecutor(AgentExecutor):
|
||||
"""Hands the question to the workflow and reports what it decided."""
|
||||
|
||||
def __init__(self, settings: Settings, http_client: httpx.AsyncClient) -> None:
|
||||
self._settings = settings
|
||||
self._http = http_client
|
||||
|
||||
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
|
||||
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
|
||||
|
||||
if context.current_task is None:
|
||||
await event_queue.enqueue_event(
|
||||
Task(
|
||||
id=context.task_id,
|
||||
context_id=context.context_id,
|
||||
status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
|
||||
)
|
||||
)
|
||||
await updater.start_work()
|
||||
|
||||
subject_token = _subject_token(context)
|
||||
if subject_token is None:
|
||||
# Refused rather than answered. Without the end user's token the
|
||||
# workflow has no identity to act under, and an answer produced
|
||||
# anyway would be one nobody is accountable for.
|
||||
await updater.failed(
|
||||
message=updater.new_agent_message(
|
||||
parts=[Part(text="No bearer token on the request; the workflow cannot act.")]
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
response = await self._http.post(
|
||||
self._settings.webhook_url,
|
||||
headers={WORKFLOW_TOKEN_HEADER: subject_token},
|
||||
json={
|
||||
"question": context.get_user_input(),
|
||||
"context_id": context.context_id,
|
||||
"task_id": context.task_id,
|
||||
},
|
||||
timeout=self._settings.webhook_timeout,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 — surfaced to the caller, logged here
|
||||
logger.exception("the workflow could not be reached")
|
||||
await updater.failed(
|
||||
message=updater.new_agent_message(
|
||||
parts=[Part(text=f"The workflow could not be reached: {exc}")]
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
if response.status_code >= 400:
|
||||
# The workflow's own body is the useful part — an n8n error carries
|
||||
# which node failed, which a status code does not.
|
||||
await updater.failed(
|
||||
message=updater.new_agent_message(
|
||||
parts=[Part(text=f"The workflow failed ({response.status_code}): "
|
||||
f"{response.text[:500]}")]
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
answer, escalate = _read(response)
|
||||
if escalate:
|
||||
# An autonomous agent that meets a rule needing approval must fail
|
||||
# cleanly and escalate, never decide. The caller sees a task that
|
||||
# did not complete, which is the honest shape for "a human has to
|
||||
# look at this".
|
||||
await updater.failed(message=updater.new_agent_message(parts=[Part(text=answer)]))
|
||||
return
|
||||
|
||||
await updater.complete(message=updater.new_agent_message(parts=[Part(text=answer)]))
|
||||
|
||||
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
|
||||
await TaskUpdater(event_queue, context.task_id, context.context_id).cancel()
|
||||
|
||||
|
||||
def _read(response: httpx.Response) -> tuple[str, bool]:
|
||||
"""What the workflow said, and whether it is escalating.
|
||||
|
||||
A workflow answers with `{"answer": "...", "escalate": false}`. Anything else
|
||||
is returned as raw text rather than guessed at — a workflow that has drifted
|
||||
from the contract should be visible, not silently reinterpreted.
|
||||
"""
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError:
|
||||
return response.text.strip() or "(the workflow returned nothing)", False
|
||||
if isinstance(payload, list) and payload:
|
||||
# n8n returns an array when the workflow's last node emits items.
|
||||
payload = payload[0]
|
||||
if isinstance(payload, dict) and "answer" in payload:
|
||||
return str(payload["answer"]), bool(payload.get("escalate", False))
|
||||
return response.text.strip() or "(the workflow returned nothing)", False
|
||||
|
||||
|
||||
def _subject_token(context: RequestContext) -> str | None:
|
||||
headers = {}
|
||||
call = context.call_context
|
||||
if call is not None:
|
||||
headers = {k.lower(): v for k, v in (call.state.get("headers") or {}).items()}
|
||||
raw = headers.get(SUBJECT_TOKEN_HEADER)
|
||||
if not raw:
|
||||
return None
|
||||
token = raw[7:] if raw[:7].lower() == "bearer " else raw
|
||||
return token.strip() or None
|
||||
|
||||
|
||||
def build_app(settings: Settings) -> Starlette:
|
||||
http_client = httpx.AsyncClient(verify=settings.verify)
|
||||
card = build_agent_card(settings)
|
||||
handler = DefaultRequestHandlerV2(
|
||||
agent_executor=WorkflowExecutor(settings, http_client),
|
||||
task_store=InMemoryTaskStore(),
|
||||
agent_card=card,
|
||||
)
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def lifespan(_app: Starlette) -> AsyncIterator[None]:
|
||||
yield
|
||||
await http_client.aclose()
|
||||
|
||||
return Starlette(
|
||||
routes=create_agent_card_routes(card)
|
||||
+ create_jsonrpc_routes(handler, "/", enable_v0_3_compat=True),
|
||||
lifespan=lifespan,
|
||||
)
|
||||
Reference in New Issue
Block a user