diff --git a/adapter/README.md b/adapter/README.md new file mode 100644 index 0000000..d388667 --- /dev/null +++ b/adapter/README.md @@ -0,0 +1,44 @@ +# a2a-n8n-adapter + +An A2A facade in front of an n8n workflow. One image, one instance per agent, +configured entirely by environment. + +The repo's 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. Templating the whole lifecycle in +gateway configuration is an adapter, written in the wrong language. + +So: the alternative those same notes anticipated, written once for both agents. + +## It owns the protocol and nothing else + +It holds no Keycloak client and no credentials. It forwards the end user's token +to the workflow in `X-Subject-Token` — a header rather than the body, so a +workflow that logs its input does not log the token with it — and the workflow +exchanges it, because the component that can mint tokens should be the one that +uses them. + +The workflow answers `{"answer": "...", "escalate": false}`. Anything else is +returned as raw text rather than guessed at: a workflow that has drifted from +the contract should look wrong, not be quietly reinterpreted. + +`escalate: true` fails the task rather than completing it. An autonomous agent +that meets a rule needing approval must hand it back, never decide, and a +completed task is the wrong shape for "a human has to look at this". + +## Configuration + +| | | +|---|---| +| `ADAPTER_AGENT_NAME` | `approvisionnement` | +| `ADAPTER_AGENT_DESCRIPTION` | what the card says it is | +| `ADAPTER_SKILL_ID` | `check_availability` | +| `ADAPTER_SKILL_NAME`, `_DESCRIPTION`, `_EXAMPLES` | the card's one skill; examples are newline-separated | +| `ADAPTER_WEBHOOK_URL` | the workflow's production webhook | +| `ADAPTER_PUBLIC_URL` | what callers dial — the Service DNS name, not the bind address | +| `ADAPTER_CA_BUNDLE` | the lab CA, so TLS is verified rather than disabled | + +One skill per instance, deliberately. A second one is a card change, which is +the moment to ask whether the agent has quietly become a data proxy. diff --git a/adapter/pyproject.toml b/adapter/pyproject.toml new file mode 100644 index 0000000..1a7418c --- /dev/null +++ b/adapter/pyproject.toml @@ -0,0 +1,26 @@ +[project] +name = "a2a-n8n-adapter" +version = "0.1.0" +description = "An A2A facade in front of an n8n workflow" +readme = "README.md" +requires-python = ">=3.12" +dependencies = [ + "a2a-sdk[http-server]>=1.1.2,<2", + "httpx>=0.28.1,<1", + "pydantic-settings>=2.7", + "uvicorn>=0.52.4", +] + +[project.optional-dependencies] +dev = ["pytest>=8", "pytest-asyncio>=0.24"] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/a2a_n8n_adapter"] + +[tool.pytest.ini_options] +pythonpath = ["src"] +asyncio_mode = "auto" diff --git a/adapter/src/a2a_n8n_adapter/__init__.py b/adapter/src/a2a_n8n_adapter/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/adapter/src/a2a_n8n_adapter/__main__.py b/adapter/src/a2a_n8n_adapter/__main__.py new file mode 100644 index 0000000..131e8d6 --- /dev/null +++ b/adapter/src/a2a_n8n_adapter/__main__.py @@ -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() diff --git a/adapter/src/a2a_n8n_adapter/__pycache__/__init__.cpython-312.pyc b/adapter/src/a2a_n8n_adapter/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..1396f46 Binary files /dev/null and b/adapter/src/a2a_n8n_adapter/__pycache__/__init__.cpython-312.pyc differ diff --git a/adapter/src/a2a_n8n_adapter/__pycache__/config.cpython-312.pyc b/adapter/src/a2a_n8n_adapter/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000..30ab8d4 Binary files /dev/null and b/adapter/src/a2a_n8n_adapter/__pycache__/config.cpython-312.pyc differ diff --git a/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc b/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc new file mode 100644 index 0000000..44b643c Binary files /dev/null and b/adapter/src/a2a_n8n_adapter/__pycache__/server.cpython-312.pyc differ diff --git a/adapter/src/a2a_n8n_adapter/config.py b/adapter/src/a2a_n8n_adapter/config.py new file mode 100644 index 0000000..1bf8ac1 --- /dev/null +++ b/adapter/src/a2a_n8n_adapter/config.py @@ -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()] diff --git a/adapter/src/a2a_n8n_adapter/server.py b/adapter/src/a2a_n8n_adapter/server.py new file mode 100644 index 0000000..ceb774c --- /dev/null +++ b/adapter/src/a2a_n8n_adapter/server.py @@ -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, + ) diff --git a/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc b/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc new file mode 100644 index 0000000..a9816cb Binary files /dev/null and b/adapter/tests/__pycache__/test_adapter.cpython-312-pytest-9.1.1.pyc differ diff --git a/adapter/tests/test_adapter.py b/adapter/tests/test_adapter.py new file mode 100644 index 0000000..5e88d6a --- /dev/null +++ b/adapter/tests/test_adapter.py @@ -0,0 +1,96 @@ +"""What the adapter must do regardless of any workflow. + +The adapter owns the protocol and nothing else, so what is worth testing is the +protocol edges: a token that is absent or oddly cased, a workflow that answers +off-contract, and the escalation flag that must not be silently swallowed. +""" + +from __future__ import annotations + +import httpx +import pytest + +from a2a_n8n_adapter.config import Settings +from a2a_n8n_adapter.server import _read, build_agent_card + + +def settings(**over) -> Settings: + base = dict( + agent_name="approvisionnement", + agent_description="Supply decisions.", + skill_id="check_availability", + skill_name="Check availability", + skill_description="Can this be promised?", + skill_examples="Can we promise S10_1949 by 15 March?\nIs S18_3782 running out?", + webhook_url="http://n8n/webhook/x", + ) + return Settings(**{**base, **over}) + + +def response(payload, status=200) -> httpx.Response: + if isinstance(payload, str): + return httpx.Response(status, text=payload) + return httpx.Response(status, json=payload) + + +class TestReadingTheWorkflow: + def test_the_contracted_shape(self) -> None: + assert _read(response({"answer": "yes", "escalate": False})) == ("yes", False) + + def test_n8n_wraps_its_last_node_in_an_array(self) -> None: + assert _read(response([{"answer": "yes"}])) == ("yes", False) + + def test_escalation_is_carried_not_swallowed(self) -> None: + # An autonomous agent meeting a rule that needs approval must escalate. + # Losing this flag would turn "a human must look" into "done". + assert _read(response({"answer": "needs approval", "escalate": True})) == ( + "needs approval", + True, + ) + + def test_an_off_contract_reply_is_returned_raw(self) -> None: + # Visible rather than reinterpreted: a workflow that has drifted should + # look wrong, not be quietly guessed at. + text, escalate = _read(response("plain text")) + assert text == "plain text" and escalate is False + + def test_an_empty_reply_says_so(self) -> None: + assert _read(response("")) == ("(the workflow returned nothing)", False) + + +class TestAgentCard: + def test_it_advertises_exactly_one_skill(self) -> None: + # One skill per adapter instance. A second one is a card change, which + # is the moment to ask whether the agent has become a data proxy. + card = build_agent_card(settings()) + assert len(card.skills) == 1 + assert card.skills[0].id == "check_availability" + + def test_examples_reach_the_card(self) -> None: + # A calling agent's model reads these to decide whether to ask at all. + card = build_agent_card(settings()) + assert len(card.skills[0].examples) == 2 + + def test_the_card_advertises_the_public_url_not_the_bind_address(self) -> None: + card = build_agent_card(settings(public_url="http://approvisionnement.agents:8080/")) + assert card.supported_interfaces[0].url == "http://approvisionnement.agents:8080/" + + +class TestSubjectToken: + @pytest.mark.parametrize("raw", ["Bearer abc", "bearer abc", "BEARER abc", "abc"]) + def test_the_prefix_is_optional_and_case_insensitive(self, raw: str) -> None: + from a2a_n8n_adapter.server import _subject_token + + class Ctx: + class call_context: # noqa: N801 + state = {"headers": {"Authorization": raw}} + + assert _subject_token(Ctx()) == "abc" + + def test_an_absent_token_is_none_rather_than_an_empty_string(self) -> None: + from a2a_n8n_adapter.server import _subject_token + + class Ctx: + call_context = None + + assert _subject_token(Ctx()) is None diff --git a/workflows/README.md b/workflows/README.md new file mode 100644 index 0000000..11d979f --- /dev/null +++ b/workflows/README.md @@ -0,0 +1,48 @@ +# Workflows + +Exported as JSON and committed. The n8n instance is not the source of truth. + +## Importing + +n8n's REST API, with an owner session cookie. The activation endpoint is +separate from the update one and requires the current `versionId`: + +``` +POST /rest/workflows # import +POST /rest/workflows//activate # {"versionId": ""} +``` + +`PATCH /rest/workflows/` with `{"active": true}` returns 200 and leaves the +workflow inactive, silently. Use the activate endpoint. + +## approvisionnement-check-availability + +**Not working yet.** The `Exchange token` node fails, and what it fails with is +worth recording: + +``` +Keycloak 500, an HTML login page as the body +NullPointerException: Cannot invoke UserSessionModel.getAuthenticatedClientSessionByClient + because "userSession" is null + at V1TokenExchangeProvider.exchangeClientToOIDCClient +``` + +Two things stand out. Keycloak answers a malformed token exchange with a **500 +and a rendered login page**, which tells a client nothing. And it took the **V1** +exchange path, not the Standard one every other agent in this estate uses +successfully. + +What has been eliminated, each by hand against the same running Keycloak: + +- the client and its secret — Keycloak logs + `kc.clientId: classic-models-agent-approvisionnement`, so it authenticated, + which also proves `$env` reaches the node +- the subject token — the *same* token replayed with curl exchanges fine (200) +- omitting `subject_token_type` — still 200 +- omitting `openid` from the scope — still 200 +- using `audience=` instead of a scope — a clean 400, not this + +So n8n is sending something the reproduction attempts are not. The next step is +to capture the request as the node actually emits it: point the node at an +in-cluster echo service and read the body. That was set up and the run did not +reach it, which is itself unexplained and where this stopped. diff --git a/workflows/approvisionnement-check-availability.json b/workflows/approvisionnement-check-availability.json new file mode 100644 index 0000000..7d9f6a6 --- /dev/null +++ b/workflows/approvisionnement-check-availability.json @@ -0,0 +1,257 @@ +{ + "name": "Approvisionnement \u2014 check_availability", + "nodes": [ + { + "parameters": { + "httpMethod": "POST", + "path": "check-availability", + "responseMode": "responseNode", + "options": {} + }, + "id": "webhook", + "name": "Webhook", + "type": "n8n-nodes-base.webhook", + "typeVersion": 2, + "position": [ + 0, + 0 + ], + "webhookId": "check-availability" + }, + { + "parameters": { + "method": "POST", + "url": "https://keycloak.sttlab.local/realms/classic-models/protocol/openid-connect/token", + "sendBody": true, + "contentType": "form-urlencoded", + "bodyParameters": { + "parameters": [ + { + "name": "grant_type", + "value": "urn:ietf:params:oauth:grant-type:token-exchange" + }, + { + "name": "client_id", + "value": "classic-models-agent-approvisionnement" + }, + { + "name": "client_secret", + "value": "={{ $env.APPROVISIONNEMENT_CLIENT_SECRET }}" + }, + { + "name": "subject_token", + "value": "={{ $('Webhook').item.json.headers['x-subject-token'] }}" + }, + { + "name": "subject_token_type", + "value": "urn:ietf:params:oauth:token-type:access_token" + }, + { + "name": "scope", + "value": "openid mcp-catalog-audience token-exchange-broker-audience" + } + ] + }, + "options": {} + }, + "id": "exchange-token", + "name": "Exchange token", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [ + 220, + 0 + ] + }, + { + "parameters": { + "method": "POST", + "url": "https://catalog-dev.sttlab.local/catalog-mcp", + "sendHeaders": true, + "headerParameters": { + "parameters": [ + { + "name": "Authorization", + "value": "=Bearer {{ $json.access_token }}" + }, + { + "name": "Accept", + "value": "application/json, text/event-stream" + } + ] + }, + "sendBody": true, + "specifyBody": "json", + "jsonBody": "{\"jsonrpc\": \"2.0\", \"id\": 1, \"method\": \"initialize\", \"params\": {\"protocolVersion\": \"2025-06-18\", \"capabilities\": {}, \"clientInfo\": {\"name\": \"approvisionnement\", \"version\": \"0.1.0\"}}}", + "options": { + "response": { + "response": { + "fullResponse": true + } + } + } + }, + "id": "mcp-initialize", + "name": "MCP initialize", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [ + 440, + 0 + ] + }, + { + "parameters": { + "method": "POST", + "url": "https://catalog-dev.sttlab.local/catalog-mcp", + "sendHeaders": true, + "headerParameters": { + "parameters": [ + { + "name": "Authorization", + "value": "=Bearer {{ $('Exchange token').item.json.access_token }}" + }, + { + "name": "Accept", + "value": "application/json, text/event-stream" + }, + { + "name": "mcp-session-id", + "value": "={{ $json.headers['mcp-session-id'] }}" + } + ] + }, + "sendBody": true, + "specifyBody": "json", + "jsonBody": "={{ JSON.stringify({ jsonrpc: '2.0', id: 2, method: 'tools/call', params: { name: 'get-product', arguments: { path_productcode: $('Decide the product').item.json.productcode } } }) }}", + "options": { + "response": { + "response": { + "neverError": true + } + } + } + }, + "id": "read-stock", + "name": "Read stock", + "type": "n8n-nodes-base.httpRequest", + "typeVersion": 4.2, + "position": [ + 660, + 0 + ] + }, + { + "parameters": { + "jsCode": "// Pull the product code out of the question.\n//\n// A regex rather than a model: this agent's critical path is rules, and putting\n// an LLM on it would make a nightly job both slower and less predictable. If the\n// question does not name a product, the agent escalates instead of guessing \u2014\n// an autonomous agent that meets an ambiguity must hand it back, never resolve\n// it on its own.\nconst question = $('Webhook').first().json.body?.question ?? '';\nconst match = question.match(/\\bS\\d+_\\d+[A-Za-z]*\\b/);\nif (!match) {\n return [{ json: { productcode: null, question } }];\n}\nreturn [{ json: { productcode: match[0], question } }];\n" + }, + "id": "decide-the-product", + "name": "Decide the product", + "type": "n8n-nodes-base.code", + "typeVersion": 2, + "position": [ + 220, + 180 + ] + }, + { + "parameters": { + "jsCode": "// Turn the MCP reply into a decision.\n//\n// An A2A call returns an answer, not data: the caller learns whether the stock\n// covers what it asked about, and never receives the stock table. That is what\n// keeps this agent from becoming a read proxy onto the catalog domain.\nconst THRESHOLD = 50; // below this, a reorder is due before promising anything\n\nconst decided = $('Decide the product').first().json;\nif (!decided.productcode) {\n return [{ json: {\n answer: 'No product reference in the question. Name one as S10_1949 and ask again.',\n escalate: true,\n }}];\n}\n\nconst raw = $input.first().json;\nconst body = typeof raw === 'string' ? raw : (raw.data ?? JSON.stringify(raw));\n\n// The endpoint answers as an SSE event, so the JSON sits on a `data:` line.\nconst line = String(body).split('\\n').find((l) => l.startsWith('data:'));\nif (!line) {\n return [{ json: {\n answer: `Could not read the catalog's reply for ${decided.productcode}.`,\n escalate: true,\n }}];\n}\n\nconst rpc = JSON.parse(line.slice(5).trim());\nif (rpc.error) {\n // A refusal is the caller's own permissions being applied. Reported, never\n // worked around.\n return [{ json: {\n answer: `The catalog refused the read for ${decided.productcode}: ${rpc.error.message}.`,\n escalate: true,\n }}];\n}\n\nconst product = JSON.parse(rpc.result.content[0].text);\nconst stock = Number(product.quantityinstock);\nconst available = stock >= THRESHOLD;\n\nreturn [{ json: {\n answer: available\n ? `Yes \u2014 ${product.productname} (${product.productcode}) can be promised: `\n + `${stock} in stock, above the reorder threshold of ${THRESHOLD}.`\n : `No \u2014 ${product.productname} (${product.productcode}) is down to ${stock}, `\n + `below the reorder threshold of ${THRESHOLD}. A reorder is due before it can be promised.`,\n escalate: false,\n}}];\n" + }, + "id": "answer", + "name": "Answer", + "type": "n8n-nodes-base.code", + "typeVersion": 2, + "position": [ + 880, + 0 + ] + }, + { + "parameters": { + "respondWith": "json", + "responseBody": "={{ JSON.stringify($json) }}", + "options": {} + }, + "id": "respond", + "name": "Respond", + "type": "n8n-nodes-base.respondToWebhook", + "typeVersion": 1.1, + "position": [ + 1100, + 0 + ] + } + ], + "connections": { + "Webhook": { + "main": [ + [ + { + "node": "Decide the product", + "type": "main", + "index": 0 + } + ] + ] + }, + "Decide the product": { + "main": [ + [ + { + "node": "Exchange token", + "type": "main", + "index": 0 + } + ] + ] + }, + "Exchange token": { + "main": [ + [ + { + "node": "MCP initialize", + "type": "main", + "index": 0 + } + ] + ] + }, + "MCP initialize": { + "main": [ + [ + { + "node": "Read stock", + "type": "main", + "index": 0 + } + ] + ] + }, + "Read stock": { + "main": [ + [ + { + "node": "Answer", + "type": "main", + "index": 0 + } + ] + ] + }, + "Answer": { + "main": [ + [ + { + "node": "Respond", + "type": "main", + "index": 0 + } + ] + ] + } + }, + "settings": { + "executionOrder": "v1" + } +}