From c893934656703015756230ec463c244a374880a3 Mon Sep 17 00:00:00 2001 From: Abhishek Singh Date: Sat, 5 Sep 2026 14:48:39 -0700 Subject: [PATCH] Added samples for A2A gateway --- gcloud/README.md | 134 +++++++++++++++++++++++---- gcloud/gateway/README.md | 9 ++ samples/README.md | 17 ++++ samples/a2a/gateway/README.md | 107 +++++++++++++++++++++ samples/a2a/gateway/_common.py | 125 +++++++++++++++++++++++++ samples/a2a/gateway/blocking_task.py | 21 +++++ samples/a2a/gateway/task_poll.py | 31 +++++++ 7 files changed, 428 insertions(+), 16 deletions(-) create mode 100644 samples/a2a/gateway/README.md create mode 100644 samples/a2a/gateway/_common.py create mode 100644 samples/a2a/gateway/blocking_task.py create mode 100644 samples/a2a/gateway/task_poll.py diff --git a/gcloud/README.md b/gcloud/README.md index cf52356..830cebc 100644 --- a/gcloud/README.md +++ b/gcloud/README.md @@ -1,19 +1,121 @@ - # Google Cloud deployment modes +# Google Cloud deployment modes -Select AI supports two distinct Google Cloud deployment modes. Choose based on -whether the database and Select AI team are known at deployment time or must -be selected dynamically by each user. +Select AI for Python supports two A2A deployment architectures. The key +decisions are where the database connection and Select AI team are selected, +which components carry the session, and how capacity is added: -| | Standalone | Dynamic gateway | +- Standalone fixes the database and team at deployment time. One Cloud Run A2A + service owns the configured connection pool and serves that team. +- The gateway selects the database and team per user session. A Cloud Run A2A + gateway routes sessions through Consul to a clustered GKE worker pool, with + one isolated child runtime and database connection pool per active session. + +The gateway architecture is designed for horizontal session capacity. Gateway +instances, Consul, and worker replicas are separate components; adding worker +replicas increases the number of concurrent database sessions that can be +hosted behind the same A2A endpoint. Consul preserves session and task affinity +when requests reach different gateway instances. Oracle Database capacity and +the configured session TTL remain the limiting factors. + +## What the A2A client connects to + +### Standalone server + +The standalone deployment is one Cloud Run A2A service for one configured +Oracle database and one Select AI team. + +```text +A2A client ── A2A JSON-RPC ──► Cloud Run A2A server ──► Oracle Database + fixed credentials + fixed team +``` + +The service receives its database credentials from Secret Manager. The A2A +client can discover the Agent Card and immediately send a database prompt. +The server supports blocking tasks, task polling, and streaming responses. + +Deploy it with: + +```bash +gcloud/standalone/deploy.sh --build +``` + +Use [standalone deployment](standalone/README.md) for the deployment details. + +### Dynamic gateway + +The gateway deployment provides one public A2A endpoint for users who choose +the database and Select AI team at runtime. + +```text +A2A client ── A2A JSON-RPC ──► Cloud Run gateway + │ A2UI connection form + ▼ + GKE worker session ──► Oracle Database + ▲ + │ Consul session/task routing +``` + +The client first sends a message and receives an A2UI connection form. After +the client submits the DSN, username, password, and team name, the gateway +opens a temporary worker session. Subsequent A2A messages use that session and +execute against the selected database and team. + +The gateway supports blocking tasks and asynchronous task polling. Its Agent +Card advertises `streaming: false`; clients use `message/send` followed by +`tasks/get` for long-running work. The gateway-to-worker path uses internal +protobuf messages, while the public client-facing path remains A2A JSON-RPC. + +The gateway database session currently accepts a DSN, username, and password. +Wallet-based Oracle Database mTLS is not yet supported by this session path. +The optional mTLS deployment mode described in the gateway documentation +secures the gateway-to-worker connection; it is separate from database mTLS. + +Deploy it with: + +```bash +gcloud/gateway/deploy.sh --project PROJECT_ID +``` + +Use [gateway deployment](gateway/README.md) for the deployment details. + +## Client-visible differences + +| Client concern | Standalone server | Dynamic gateway | | --- | --- | --- | -| Database and team | Fixed at deployment time | Chosen at runtime for each user session | -| Public A2A service | One service for one configured team | One gateway that presents an A2UI connection form | -| Users | All requests use the deployed database identity | Any permitted user can connect to a reachable Oracle database and Select AI team | -| Architecture | One Cloud Run service | Cloud Run gateway, plus Consul and worker replicas in GKE | -| Session isolation | Shared service database pool | One child process and async pool per active user session | -| Main benefit | Simple, predictable deployment | Dynamic, multi-database and multi-team access from one A2A endpoint | -| Operational cost | Low | Higher: GKE workers, Consul, routing, TTL, and session capacity | - -Use [standalone](standalone/README.md) when a service should expose one known -database team. Use [gateway](gateway/README.md) when users must dynamically -choose their database connection and team. +| Database/team selection | Configured by the deployment | Submitted by each user session through A2UI | +| First client operation | Send the database prompt | Send a prompt, submit the connection form, then send the database prompt | +| Credentials | Stored in Secret Manager for the service | Supplied for the temporary session and held by its worker | +| Database mTLS | Supported through the standalone wallet configuration | Not yet supported for gateway database sessions | +| Public service | One Cloud Run A2A service | Cloud Run gateway backed by GKE workers and Consul | +| Agent Card input | `text/plain` | `text/plain` and `application/json+a2ui` | +| Agent Card streaming | `true` | `false` | +| Blocking request | `message/send` waits for the final task result | `message/send` waits for the final task result after the session is connected | +| Streaming response | Supported through A2A streaming methods and SSE | Not available; clients use task polling | +| Asynchronous task | `message/send` with `configuration.blocking: false` | `message/send` with `configuration.blocking: false` | +| Task polling | `tasks/get` until the task reaches a terminal state | `tasks/get` until the task reaches a terminal state | +| Session ownership | Cloud Run service database pool | One child process and async pool per active user session | +| Task/context storage | Oracle Database | Oracle Database, with Consul routing metadata | +| Capacity control | Cloud Run instances and per-instance pool size | Gateway instances, Consul routing, worker replicas, per-session pools, and session TTL | +| Best fit | One known database/team and predictable operations | Multiple databases/teams selected dynamically from one endpoint | + +Both deployments expose the public A2A endpoint at: + +```text +/.well-known/agent-card.json +/a2a/jsonrpc/ +``` + +Both accept A2A 1.0 method names and the A2A v0.3 compatibility method names. +The gateway client flow is documented in the +[gateway samples](../samples/a2a/gateway/README.md). + +## Which deployment should you choose? + +Choose the standalone server when the service owner controls the database +identity and team, wants clients to send prompts immediately, and benefits +from streaming responses. + +Choose the gateway when one A2A endpoint must serve users selecting different +Oracle databases or teams, with isolated temporary sessions and worker-based +capacity. diff --git a/gcloud/gateway/README.md b/gcloud/gateway/README.md index a117207..5e11273 100644 --- a/gcloud/gateway/README.md +++ b/gcloud/gateway/README.md @@ -42,6 +42,11 @@ using the internal protobuf protocol. The selected worker starts one child runtime for that session. The child owns the database connection, `DefaultRequestHandler`, `OracleTaskStore`, and `OracleContextStore`. +The session connection path currently accepts a DSN, username, and password. +Oracle Database wallet-based mTLS is not yet supported for these dynamic +sessions. The optional worker mTLS mode below protects the gateway-to-worker +HTTP connection; it does not provide database mTLS. + The Service Registry stores only service-discovery and non-secret session/task-to-worker metadata. Task payloads and context mappings remain in Oracle. Connection-form tasks are response-only bootstrap tasks: they are @@ -130,6 +135,10 @@ deployment time. Local testing does not use mTLS. The default GCloud deployment also keeps the current private-VPC HTTP worker transport. +This mTLS mode applies only between the Cloud Run gateway and GKE workers. It +is independent of Oracle Database authentication, and does not enable wallet- +based database mTLS for gateway sessions. + For a short-lived GCloud mTLS test: ```bash diff --git a/samples/README.md b/samples/README.md index 858e66f..8e0c860 100644 --- a/samples/README.md +++ b/samples/README.md @@ -52,6 +52,23 @@ This sample intentionally omits `configuration.blocking`. The server waits for the database work to finish and returns the completed Task in the initial `message/send` response; no polling is needed. +## A2A dynamic gateway + +The dynamic gateway samples submit the A2UI database connection form, open a +temporary worker session, and execute database tasks. The gateway advertises +`streaming: false` and supports non-blocking task execution with +`configuration.blocking: false` and `tasks/get`. + +Gateway-specific samples that perform the form handshake and then execute a +real database task are in [a2a/gateway](a2a/gateway/README.md): + +```bash +python samples/a2a/gateway/blocking_task.py +python samples/a2a/gateway/task_poll.py +``` + +See that README for local Consul, worker, and gateway startup instructions. + `SELECT_AI_DB_CONNECT_STRING` can be in any one of the following formats diff --git a/samples/a2a/gateway/README.md b/samples/a2a/gateway/README.md new file mode 100644 index 0000000..c3349da --- /dev/null +++ b/samples/a2a/gateway/README.md @@ -0,0 +1,107 @@ +# Dynamic A2A gateway samples + +These samples connect to a dynamic gateway, submit its A2UI database +connection form, open a temporary worker session, and execute Select AI tasks +against the database. + +The gateway advertises `streaming: false` and supports request/response task +operations. Long-running work is returned as a task and can be followed with +`tasks/get`. + +The gateway supports asynchronous work with `message/send` and +`configuration.blocking: false`, followed by `tasks/get`. + +These samples use the A2A v0.3 JSON-RPC names used by the existing samples: +`message/send`, `tasks/get`, and `tasks/cancel`. A client using A2A 1.0 should +send `A2A-Version: 1.0` and use `SendMessage`, `GetTask`, `ListTasks`, and +`CancelTask`; its non-blocking option is `configuration.returnImmediately`. +The gateway accepts both versions, but streaming is disabled in both. + +## Local setup + +Install the A2A extra if necessary: + +```bash +source .venv/bin/activate +pip install -e '.[a2a]' +``` + +Run these commands in three terminals from the repository root. + +Terminal 1, Consul: + +```bash +consul agent -dev -bind=127.0.0.1 -client=127.0.0.1 +``` + +Terminal 2, one worker: + +```bash +source .venv/bin/activate + +CONSUL_HTTP_URL=http://127.0.0.1:8500 \ +WORKER_ID=local-worker \ +WORKER_ADDRESS=127.0.0.1 \ +WORKER_PORT=8081 \ +select-ai a2a worker --host 127.0.0.1 --port 8081 +``` + +Terminal 3, the gateway: + +```bash +source .venv/bin/activate + +select-ai a2a gateway \ + --host 127.0.0.1 \ + --port 8000 \ + --agent-url http://127.0.0.1:8000 \ + --consul-url http://127.0.0.1:8500 +``` + +The worker must be able to connect to the database when a sample submits the +form. Export the same values used by the other samples, plus the optional +team name: + +```bash +export SELECT_AI_DB_CONNECT_STRING='' +export SELECT_AI_USER='' +export SELECT_AI_PASSWORD='' +export SELECT_AI_A2A_TEAM='ORACLE_AI_DATABASE_AGENT' +``` + +For a TNS-alias DSN, set `TNS_ADMIN` in the worker terminal before starting +the worker. Wallet-based Oracle Database mTLS is not currently supported by +the gateway session connection path. + +## Run the samples + +The gateway-specific samples perform the connection-form handshake +automatically and validate that the final artifact is +`database-agent-result`. + +Blocking database task: + +```bash +python samples/a2a/gateway/blocking_task.py +``` + +Non-blocking database task with polling: + +```bash +python samples/a2a/gateway/task_poll.py +``` + +Expected task output is similar to: + +```text +Task : completed +Artifact: database-agent-result +... +``` + +For a quick health check: + +```bash +curl http://127.0.0.1:8081/health +curl -sS http://127.0.0.1:8000/.well-known/agent-card.json | jq +``` diff --git a/samples/a2a/gateway/_common.py b/samples/a2a/gateway/_common.py new file mode 100644 index 0000000..25b03a5 --- /dev/null +++ b/samples/a2a/gateway/_common.py @@ -0,0 +1,125 @@ +# ----------------------------------------------------------------------------- +# Copyright (c) 2026, Oracle and/or its affiliates. +# +# Licensed under the Universal Permissive License v 1.0 as shown at +# https://oss.oracle.com/licenses/upl. +# ----------------------------------------------------------------------------- + +"""Small A2A v0.3 client helpers for the dynamic gateway samples.""" + +import json +import os +import uuid +from urllib.request import Request, urlopen + +ENDPOINT = os.environ.get( + "SELECT_AI_A2A_GATEWAY_ENDPOINT", + "http://127.0.0.1:8000/a2a/jsonrpc/", +) +TEAM_NAME = os.environ.get("SELECT_AI_A2A_TEAM", "ORACLE_AI_DATABASE_AGENT") + + +def call(method: str, params: dict) -> dict: + """Make one A2A v0.3 JSON-RPC call and return its result.""" + request = Request( + ENDPOINT, + data=json.dumps( + { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": method, + "params": params, + } + ).encode(), + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urlopen(request) as response: # noqa: S310 + body = json.load(response) + if "error" in body: + raise RuntimeError(body["error"]) + return body["result"] + + +def _message(text: str, context_id: str | None = None) -> dict: + message = { + "messageId": str(uuid.uuid4()), + "role": "user", + "parts": [{"kind": "text", "text": text}], + } + if context_id: + message["contextId"] = context_id + return message + + +def connect(prompt: str) -> str: + """Bootstrap one gateway session and return its context ID.""" + form_task = call("message/send", {"message": _message(prompt)}) + if form_task["artifacts"][0]["name"] != "database-connection-form": + raise RuntimeError( + "Expected database-connection-form, got " + f"{form_task['artifacts'][0].get('name')}" + ) + + context_id = form_task["contextId"] + connection_task = call( + "message/send", + { + "message": { + "messageId": str(uuid.uuid4()), + "contextId": context_id, + "taskId": form_task["id"], + "role": "user", + "parts": [ + { + "kind": "data", + "data": { + "version": "v0.9", + "action": { + "name": "submit_database_connection", + "context": { + "dsn": os.environ[ + "SELECT_AI_DB_CONNECT_STRING" + ], + "username": os.environ["SELECT_AI_USER"], + "password": os.environ[ + "SELECT_AI_PASSWORD" + ], + "team_name": TEAM_NAME, + }, + }, + }, + "metadata": {"mimeType": "application/json+a2ui"}, + } + ], + } + }, + ) + artifact = connection_task["artifacts"][0] + if artifact["name"] != "database-session": + raise RuntimeError(f"Database connection failed: {artifact['name']}") + return context_id + + +def send_prompt( + prompt: str, + context_id: str, + blocking: bool | None = None, +) -> dict: + """Send a normal database prompt through an existing gateway session.""" + params = {"message": _message(prompt, context_id)} + if blocking is False: + params["configuration"] = {"blocking": False} + return call("message/send", params) + + +def print_task_summary(task: dict, *, include_task: bool = True) -> None: + """Print a useful result without dumping connection-form internals.""" + artifact = (task.get("artifacts") or [{}])[0] + parts = artifact.get("parts") or [{}] + if include_task: + print(f"Task {task['id']}: {task['status']['state']}") + print(f"Artifact: {artifact.get('name')}") + for part in parts: + if "text" in part: + print(part["text"]) diff --git a/samples/a2a/gateway/blocking_task.py b/samples/a2a/gateway/blocking_task.py new file mode 100644 index 0000000..5f18f8a --- /dev/null +++ b/samples/a2a/gateway/blocking_task.py @@ -0,0 +1,21 @@ +# ----------------------------------------------------------------------------- +# Copyright (c) 2026, Oracle and/or its affiliates. +# +# Licensed under the Universal Permissive License v 1.0 as shown at +# https://oss.oracle.com/licenses/upl. +# ----------------------------------------------------------------------------- + +"""Connect to a dynamic A2A gateway, then send a blocking database request.""" + +from _common import connect, print_task_summary, send_prompt + +PROMPT = "What were last month's sales by product category?" + + +context_id = connect(PROMPT) +task = send_prompt(PROMPT, context_id) +if task["artifacts"][0]["name"] != "database-agent-result": + raise RuntimeError( + "Expected a database result, got " f"{task['artifacts'][0]['name']}" + ) +print_task_summary(task) diff --git a/samples/a2a/gateway/task_poll.py b/samples/a2a/gateway/task_poll.py new file mode 100644 index 0000000..698611c --- /dev/null +++ b/samples/a2a/gateway/task_poll.py @@ -0,0 +1,31 @@ +# ----------------------------------------------------------------------------- +# Copyright (c) 2026, Oracle and/or its affiliates. +# +# Licensed under the Universal Permissive License v 1.0 as shown at +# https://oss.oracle.com/licenses/upl. +# ----------------------------------------------------------------------------- + +"""Connect to a dynamic A2A gateway, then poll a database task.""" + +import time + +from _common import call, connect, print_task_summary, send_prompt + +PROMPT = "What were last month's sales by product category?" +TERMINAL_STATES = {"completed", "failed", "canceled", "rejected"} + + +context_id = connect(PROMPT) +task = send_prompt(PROMPT, context_id, blocking=False) +print(f"Task {task['id']}: {task['status']['state']}") + +while task["status"]["state"] not in TERMINAL_STATES: + time.sleep(1) + task = call("tasks/get", {"id": task["id"]}) + print(f"Task {task['id']}: {task['status']['state']}") + +if task["artifacts"][0]["name"] != "database-agent-result": + raise RuntimeError( + "Expected a database result, got " f"{task['artifacts'][0]['name']}" + ) +print_task_summary(task, include_task=False)