feature: pluggable request/response adapter for integrations - #570
Draft
SandunYL wants to merge 2 commits into
Draft
feature: pluggable request/response adapter for integrations#570SandunYL wants to merge 2 commits into
SandunYL wants to merge 2 commits into
Conversation
amithad
reviewed
Aug 3, 2026
| - A given platform's inbound and outbound adapters are paired under one `integration` name (e.g. `"slack"`). | ||
| - Adapter selection follows the **house factory pattern** (`core/util/factory.py`): built-in short names resolved by `if/elif` + real imports, with a dotted-path bring-your-own branch — consistent with the guardrail / sandbox / store factories. | ||
|
|
||
| ### session_id resolution |
Member
There was a problem hiding this comment.
We should maintain the multi-channel capability. Meaning, same session could be picked up by multiple messaging platforms (theoretically)
amithad
reviewed
Aug 3, 2026
amithad
left a comment
amithad
left a comment
Member
There was a problem hiding this comment.
Design-stage review of design.md (spec-only PR, so implementation conformance was skipped). Overall this is a strong Stage 1 document: point-form and hierarchical, a single clarifying diagram, explicit non-goals, and staging is correct (design.md first, per the spec process).
Verification note (positive): every path:line citation in the document was checked against develop and all of them verify, including the ChatService validation asymmetry (:488 vs :366,401,450), the endpoint_url special-casing (serverless/akagentrunner.py:53-54,83-85), the enqueue contract (queue_request_handler.py:70-71,76,83-88), the six per-platform session-key derivations, and the missing Teams config section.
Findings: 2 suggestions, 5 questions, posted inline. The most important one is the Teams configuration inconsistency (the Configuration requirement cannot be satisfied for Teams as written).
Un-anchorable / PR hygiene:
- Per the spec-writing guidance, a spec-only PR should carry a
docs:title (e.g.docs: add design spec for #524 pluggable integration adapter). The current titlefeature:makes reviewers expect code, andfeatureis also not one of the house conventional-commit types (feat,fix,docs, ...). - The PR template is unfilled: no description, no type-of-change checked, and
Fixes #/Relates to #are empty. Please link issue #524, mark Documentation update (or Other: design spec), and state that implementation follows in a separate PR; mark the testing checklist items as not applicable for a spec-only PR.
|
|
||
| ### Configuration | ||
|
|
||
| - Each integration keeps its existing config section (`core/config.py:602-607`); the adapter reads the agent name and platform credentials from it. |
Member
There was a problem hiding this comment.
[suggestion] The Configuration requirement cannot be satisfied for Teams as written, and the design never resolves the gap its own Motivation calls out.
- Line 29 correctly notes Teams has no config section, yet this requirement says each integration "keeps its existing config section" and "reads the agent name and platform credentials from it". There is no
_TeamsConfigincore/config.py, andAKConfigusesextra="ignore"(core/util/config_yaml_util.py:189), so a user'steams:YAML block is dropped andteams_chat.py:40-44(Config.get().teams.agentetc.) raisesAttributeError. - Since the Migration section commits Teams to migrating in this change, the design needs an explicit requirement to add a typed
_TeamsConfig(agent, app_id, app_password, tenant_id, agent_acknowledgement), per the messaging-integration checklist (config class inconfig.py). - Suggestion: change this point to "each integration keeps its existing config section; Teams gains a new
_TeamsConfigsection (it has none today, see Motivation)".
|
|
||
| - Integrations are **queue-only with async delivery**: | ||
| - The inbound adapter verifies, normalizes, enqueues to the Input Queue, and acks the platform (HTTP 200) immediately — it does not wait for the agent. | ||
| - **Every integration reply is dual-written**: delivered to the platform via the outbound adapter, **and** written to the Response Store. This gives an audit trail / GET-polling fallback for every integration reply, at the cost of one store write per reply. This is a deliberate divergence from the WebSocket async path, which does not write to the store. |
Member
There was a problem hiding this comment.
[suggestion] The error path is unspecified: the design only covers successful replies.
- Today every synchronous handler surfaces failures to the platform user in-band (e.g. WhatsApp
whatsapp_chat.py:288, Telegramtelegram_chat.py:193, Messengermessenger_chat.py:205all send a "Sorry, ..." message), and the runner already builds error bodies for the output queue (serverless/akagentrunner.py:60_construct_error_message_body). - Under the new path, state as requirements: (1) whether agent-run errors are delivered to the platform via the outbound adapter (parity with today) or only stored; (2) what happens when outbound delivery itself fails (SQS retry / DLQ semantics on the Output Queue, and whether the store write still happens).
- These are behavioral requirements a reviewer must be able to check the implementation against, not spec.md detail; dropping the in-band error reply would be a silent user-visible behavior change.
| ### Migration | ||
|
|
||
| - The adapter path **replaces** the current synchronous handlers for the 6 webhook-based integrations — it is not an additional opt-in mode. | ||
| - Each `Agent<Platform>RequestHandler`'s inline parse/session/send logic is migrated into that platform's inbound + outbound adapter pair; the synchronous in-handler `AgentService.run_multi(...)` call inside the webhook handler is removed once the adapter path covers the platform. |
Member
There was a problem hiding this comment.
[question] Where do the mid-flight platform UX behaviors land, or are they dropped?
- Today's handlers do more than parse/run/send: WhatsApp sends an acknowledgement message before the run (
whatsapp_chat.py:283), Messenger and Instagram send mark-seen and typing indicators (messenger_chat.py:196-199), Teams hasagent_acknowledgement(teams_chat.py:41). - In the split model these can only happen at the inbound edge (after enqueue, before the reply exists). The migration section moves "parse/session/send" but is silent on these; dropping typing indicators or acks is a user-visible behavior change that should be an explicit decision either way.
| - **Every integration reply is dual-written**: delivered to the platform via the outbound adapter, **and** written to the Response Store. This gives an audit trail / GET-polling fallback for every integration reply, at the cost of one store write per reply. This is a deliberate divergence from the WebSocket async path, which does not write to the store. | ||
| - The **reply-to context rides with the message as individual flat SQS custom attributes**, not a combined JSON blob (e.g. Slack: `channel`, `thread_ts`; WhatsApp: `to_number`), alongside the existing `request_id`/`user_id`. Each integration's inbound adapter declares the small set of named attributes its outbound adapter needs, plus an `integration` name attribute identifying which adapter pair handles the message. | ||
| - **The Agent Runner forwards `integration` and the declared reply-to attributes generically and opaquely**, for any attribute set an inbound adapter attaches — not hardcoded per platform. Today only `endpoint_url` gets this treatment (`serverless/akagentrunner.py:83-85`); any other reply-routing attribute currently requires subclassing the runner per integration. After this change, no runner subclassing is needed to add a new integration. Its `ChatService` contract is unchanged. | ||
| - **The Response Handler / `ECSOutputConsumer` gains a built-in adapter-dispatch lookup** keyed by the `integration` attribute, generalizing `_broadcast_via_websocket`: no per-product subclass should be needed to add a new integration — the outbound adapter is resolved from the registry and handed the reply payload + reply-to attributes, then the reply is also written to the Response Store (see dual-write above). |
Member
There was a problem hiding this comment.
[question] How does adapter dispatch interact with the global execution.mode branch?
ResponseHandler.process_messagecurrently routes onAKConfig.get().execution.mode(serverless/akresponsehandler.py:106-112): ASYNC/STREAM broadcast, everything else goes to the store. A deployment hosting REST clients plus integrations will see both message kinds on one Output Queue.- The dispatch lookup keyed by the
integrationattribute implies per-message routing, but the precedence rule is never stated. Suggest adding one point: a message carrying anintegrationattribute is dispatched to its outbound adapter (plus the dual store write) regardless ofexecution.mode; a message without it keeps today's mode-based behavior.
| - The **reply-to context rides with the message as individual flat SQS custom attributes**, not a combined JSON blob (e.g. Slack: `channel`, `thread_ts`; WhatsApp: `to_number`), alongside the existing `request_id`/`user_id`. Each integration's inbound adapter declares the small set of named attributes its outbound adapter needs, plus an `integration` name attribute identifying which adapter pair handles the message. | ||
| - **The Agent Runner forwards `integration` and the declared reply-to attributes generically and opaquely**, for any attribute set an inbound adapter attaches — not hardcoded per platform. Today only `endpoint_url` gets this treatment (`serverless/akagentrunner.py:83-85`); any other reply-routing attribute currently requires subclassing the runner per integration. After this change, no runner subclassing is needed to add a new integration. Its `ChatService` contract is unchanged. | ||
| - **The Response Handler / `ECSOutputConsumer` gains a built-in adapter-dispatch lookup** keyed by the `integration` attribute, generalizing `_broadcast_via_websocket`: no per-product subclass should be needed to add a new integration — the outbound adapter is resolved from the registry and handed the reply payload + reply-to attributes, then the reply is also written to the Response Store (see dual-write above). | ||
| - The enqueue core currently inside `QueueRequestHandler.get_router` is factored out so both the generic `/api/v1/chat` route and integration inbound adapters share one enqueue path, instead of each integration hand-rolling its own SQS send. |
Member
There was a problem hiding this comment.
[question] Where does the factored-out enqueue core live, and in which direction does the import go?
- The enqueue core currently sits in
deployment/common/queue_request_handler.py, and adapters are placed "underintegration/or a newintegration/adapter/package" (line 53). Nointegration/module importsdeployment/today, so either the core moves somewhere both can depend on, orintegration/adapter/starts importingdeployment/common, a new lateral coupling between two peer packages. - This also decides which optional-dependency extras an integration adapter pulls in (boto3 etc.). Worth one explicit point in the design; the architecture skill only pins the core-never-imports-them rule, so this placement is a genuine open decision rather than something spec.md can silently pick.
| - Verify the raw platform event (signature / challenge) — it must receive the raw request (headers + body), since verification needs them. | ||
| - Parse the event into a `BaseRunRequest`, mapping platform identifiers into the model's **standard fields** (`prompt`, `attachments`, `agent` from the integration's config, `user_id`) rather than inventing ad-hoc extra fields (e.g. map Slack's user into `user_id`, not a custom `slack_user_id`). | ||
| - Derive the `session_id` (see session_id section). | ||
| - Derive the `request_id`: **prefer the platform's own idempotency identifier when the platform provides one** (e.g. Slack `event_id` / `client_msg_id`), falling back to a minted `uuid4` only when no such identifier exists. This makes SQS FIFO dedup collapse platform-level webhook retries instead of causing duplicate agent runs. |
Member
There was a problem hiding this comment.
[question] SQS FIFO deduplication only spans a 5-minute window; is the residual duplicate risk accepted?
- The stated rationale is that platform-native
request_ids make FIFO dedup collapse webhook retries. That holds only within SQS's fixed 5-minute dedup interval; Slack's retry schedule reaches ~5 minutes and Meta platforms retry with backoff for much longer, so a late retry can still trigger a duplicate agent run. - Fine to accept (fast 200-acks make late retries rare), but the design should state the window limitation so the claim "collapse platform-level webhook retries" is scoped honestly, per the absolute-claims rule for specs.
|
|
||
| - Define an **inbound adapter** responsibility set (framework-agnostic, under `integration/` or a new `integration/adapter/` package; core must not import it): | ||
| - Verify the raw platform event (signature / challenge) — it must receive the raw request (headers + body), since verification needs them. | ||
| - Parse the event into a `BaseRunRequest`, mapping platform identifiers into the model's **standard fields** (`prompt`, `attachments`, `agent` from the integration's config, `user_id`) rather than inventing ad-hoc extra fields (e.g. map Slack's user into `user_id`, not a custom `slack_user_id`). |
Member
There was a problem hiding this comment.
[question] Do attachment payloads fit through the queue, and should the design say so?
- Mapping attachments into
BaseRunRequest(files/images, base64) means media now rides the SQS message body (queue_request_handler.py:85enqueuesbody.model_dump()), which is capped by the SQS message size limit. Today's synchronous handlers pass downloaded binaries in-process with no such bound, and platform media (e.g. WhatsApp documents) can far exceed it. - REST queue clients already live with this constraint, but for integrations it is a new, silent behavior change. Suggest either stating the size limit as an accepted constraint (with the oversize failure mode) or noting the mitigation (e.g. store via the multimodal attachment store at the inbound edge and enqueue a reference) as a design decision or non-goal.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Type of Change
Related Issues
Fixes #
Relates to #
Changes Made
Testing
Checklist
Screenshots (if applicable)
Additional Notes