Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e4eca70c46
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| async def send(self, item: SessionMessage, /) -> None: | ||
| if self.closed: | ||
| raise anyio.ClosedResourceError | ||
| payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode() |
There was a problem hiding this comment.
For a valid 2025-11-25 task response with unlimited retention, the required Task.ttl field is explicitly None; exclude_none=True removes that nested field from the MQTT payload. Because ttl has no default, the receiver then rejects the response as a validation error instead of completing the task request. Serialize with exclude_unset=True, as the built-in stream transports do, so explicitly supplied nulls remain on the wire.
Useful? React with 👍 / 👎.
| await client.subscribe( | ||
| incoming_topic, options=SubscribeOptions(qos=2, retainAsPublished=True, retainHandling=2) | ||
| ) |
There was a problem hiding this comment.
When a broker refuses this subscription or grants QoS 0/1 instead of the requested QoS 2, MQTT 5 reports that outcome in the SUBACK reason codes returned by aiomqtt.Client.subscribe() rather than necessarily raising. Discarding the result lets the transport open even though a denial yields no inbound messages, while a lower grant makes read_messages() reject every delivery as non-QoS-2. Inspect the returned reason codes and fail entry unless the exact subscription was granted at QoS 2.
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
3 issues found across 12 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="examples/transports/demo_mqtt.py">
<violation number="1" location="examples/transports/demo_mqtt.py:28">
P1: When more than 256 messages arrive before `mqtt_transport` drains the queue, aiomqtt can drop them silently, so MCP requests disappear and wait for client timeouts. Replace this drop-on-full configuration with an overflow path that applies backpressure or fails the connection explicitly.</violation>
</file>
<file name="examples/transports/mcp_transport_examples/mqtt.py">
<violation number="1" location="examples/transports/mcp_transport_examples/mqtt.py:112">
P1: Use `exclude_unset=True` instead of `exclude_none=True` here. An explicitly supplied `Task.ttl=None` denotes unlimited retention; omitting that required field makes the peer reject an otherwise valid task response.</violation>
<violation number="2" location="examples/transports/mcp_transport_examples/mqtt.py:115">
P1: When the broker rejects this QoS-2 publish, `aiomqtt.Client.publish` still completes without exposing the negative acknowledgment, so the transport reports success and the peer can wait indefinitely for a message that was never delivered. Use an acknowledgment-aware publish path and fail the transport on non-success PUBACK/PUBCOMP reason codes.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| protocol=aiomqtt.ProtocolVersion.V5, | ||
| keepalive=15, | ||
| will=aiomqtt.Will(f"{topic}/{outgoing}", payload=b"", qos=2, retain=False), | ||
| max_queued_incoming_messages=256, |
There was a problem hiding this comment.
P1: When more than 256 messages arrive before mqtt_transport drains the queue, aiomqtt can drop them silently, so MCP requests disappear and wait for client timeouts. Replace this drop-on-full configuration with an overflow path that applies backpressure or fails the connection explicitly.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At examples/transports/demo_mqtt.py, line 28:
<comment>When more than 256 messages arrive before `mqtt_transport` drains the queue, aiomqtt can drop them silently, so MCP requests disappear and wait for client timeouts. Replace this drop-on-full configuration with an overflow path that applies backpressure or fails the connection explicitly.</comment>
<file context>
@@ -0,0 +1,42 @@
+ protocol=aiomqtt.ProtocolVersion.V5,
+ keepalive=15,
+ will=aiomqtt.Will(f"{topic}/{outgoing}", payload=b"", qos=2, retain=False),
+ max_queued_incoming_messages=256,
+ )
+ )
</file context>
| payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode() | ||
| if len(payload) > self.max_message_size: | ||
| raise ValueError("Encoded MCP message exceeds max_message_size") | ||
| await self.client.publish(self.topic, payload, qos=2, retain=False, properties=self.properties) |
There was a problem hiding this comment.
P1: When the broker rejects this QoS-2 publish, aiomqtt.Client.publish still completes without exposing the negative acknowledgment, so the transport reports success and the peer can wait indefinitely for a message that was never delivered. Use an acknowledgment-aware publish path and fail the transport on non-success PUBACK/PUBCOMP reason codes.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At examples/transports/mcp_transport_examples/mqtt.py, line 115:
<comment>When the broker rejects this QoS-2 publish, `aiomqtt.Client.publish` still completes without exposing the negative acknowledgment, so the transport reports success and the peer can wait indefinitely for a message that was never delivered. Use an acknowledgment-aware publish path and fail the transport on non-success PUBACK/PUBCOMP reason codes.</comment>
<file context>
@@ -0,0 +1,129 @@
+ payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode()
+ if len(payload) > self.max_message_size:
+ raise ValueError("Encoded MCP message exceeds max_message_size")
+ await self.client.publish(self.topic, payload, qos=2, retain=False, properties=self.properties)
+
+ async def aclose(self) -> None:
</file context>
| async def send(self, item: SessionMessage, /) -> None: | ||
| if self.closed: | ||
| raise anyio.ClosedResourceError | ||
| payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode() |
There was a problem hiding this comment.
P1: Use exclude_unset=True instead of exclude_none=True here. An explicitly supplied Task.ttl=None denotes unlimited retention; omitting that required field makes the peer reject an otherwise valid task response.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At examples/transports/mcp_transport_examples/mqtt.py, line 112:
<comment>Use `exclude_unset=True` instead of `exclude_none=True` here. An explicitly supplied `Task.ttl=None` denotes unlimited retention; omitting that required field makes the peer reject an otherwise valid task response.</comment>
<file context>
@@ -0,0 +1,129 @@
+ async def send(self, item: SessionMessage, /) -> None:
+ if self.closed:
+ raise anyio.ClosedResourceError
+ payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode()
+ if len(payload) > self.max_message_size:
+ raise ValueError("Encoded MCP message exceeds max_message_size")
</file context>
| payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode() | |
| payload = item.message.model_dump_json(by_alias=True, exclude_unset=True).encode() |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| payload = item.message.model_dump_json(by_alias=True, exclude_none=True).encode() | ||
| if len(payload) > self.max_message_size: | ||
| raise ValueError("Encoded MCP message exceeds max_message_size") | ||
| await self.client.publish(self.topic, payload, qos=2, retain=False, properties=self.properties) |
There was a problem hiding this comment.
🔴 Server operators get an error traceback, and SDK callers get raw aiomqtt exceptions, whenever the broker link drops while a write is pending. _MQTTWriter.send at mqtt.py:115 lets aiomqtt.MqttError escape, but the dispatcher only recognises BrokenResourceError/ClosedResourceError as teardown (jsonrpc_dispatcher.py:400, 474, 785, 791). Fix: in send, catch aiomqtt.MqttError from client.publish, mark the writer closed and raise anyio.BrokenResourceError from it, so every SDK write site sees the documented teardown signal. Then call_tool raises MCPError(CONNECTION_CLOSED), notify drops quietly, and the run loop ends without an exception group.
Extended reasoning...
The mqtt.py docstring lists aiomqtt.MqttError for publication failures, but the SDK layers above the writer were never written for that type. demo_mqtt_disconnect.py exercises the path: server-alice is evicted while hold is in flight. The server reader gets MqttError at mqtt.py:80, closes send, and the dispatcher run loop sees EOF at jsonrpc_dispatcher.py:512-519 and cancels the task group. _handle_request catches the cancel at 739 and calls _final_write(_write_error(CONNECTION_CLOSED)) at 746. _write_error at 790 calls _MQTTWriter.send; closed is still False because _write_stream.aclose() only runs after the task-group join at 494. client.publish on the disconnected paho socket returns MQTT_ERR_NO_CONN and aiomqtt raises MqttCodeError. Only Broken/ClosedResourceError are caught at 791, so it propagates out of the cancel branch, the task group raises an ExceptionGroup, run() fails, and runtime.py:154-157 logs "Transport connection failed" with a traceback on every such loss. The same escape hits notify (474: ctx.notify/progress in a handler raises instead of…
Verification: normal — triggered whenever the MQTT link drops (broker eviction, keepalive loss) while the SDK still has a write to perform. Mechanism verified: _MQTTWriter.send (examples/transports/mcp_transport_examples/mqtt.py:109-115) only raises ClosedResourceError when self.closed is already True, which happens solely in aclose() (line 117-121) after the transport context exits; otherwise it…
| client_transport = await open_transport(stack, "alice", session, False) | ||
| runtime = await stack.enter_async_context(server.serve()) | ||
| await runtime.connect(server_transport) | ||
| client = await stack.enter_async_context(Client(client_transport, read_timeout_seconds=None)) | ||
|
|
There was a problem hiding this comment.
🔴 CI operators get a red "Check live MQTT routing and peer loss" step after merging, if aiomqtt re-raises an unexpected disconnect on client exit as aiomqtt 2.x does. The evicted server-side aiomqtt.Client is entered on the shared stack at demo_mqtt.py:19, and after the takeover its __aexit__ raises MqttCodeError for reason 0x8E when the stack unwinds at demo_mqtt_disconnect.py:32. Fix: the demo must tolerate the deliberately evicted client's exit, e.g. close the server's broker client outside the stack under suppress(aiomqtt.MqttError), or enter it through a wrapper that swallows the eviction error, while still letting the transport's own shielded unsubscribe/close run. [also at: examples/transports/demo_mqtt_disconnect.py:33 - If aiomqtt re-raises an unexpected disconnect when a client context exits, the new CI peer-loss step fails after its assertion already passed.]
Extended reasoning...
The disconnect check needs the server-alice connection to be killed by the broker so the Last Will fires. demo_mqtt_disconnect.py:59-66 opens a second connection with username server-alice; with use_username_as_clientid true the broker takes over the session and sends the old connection a DISCONNECT with reason 0x8E. paho invokes on_disconnect with that reason code; aiomqtt's _on_disconnect sets the client's _disconnected future to an exception because the reason is not MQTT_ERR_SUCCESS. The client's messages iterator then raises MqttError, which mqtt.py:81 catches, so the read stream ends and the pending call fails with CONNECTION_CLOSED as asserted. finished.wait() returns and the task group exits. AsyncExitStack then unwinds: MCP Client, then runtime, then the client-side aiomqtt.Client (graceful), then the server-side aiomqtt.Client entered at demo_mqtt.py:19. aiomqtt 2.x aexit checks _disconnected.done() and, when it holds an exception, re-raises it. That MqttCodeError propagates out of…
Verification: normal — triggering condition: the pinned aiomqtt (uv.lock: name = "aiomqtt" / version = "2.5.1") keeps the aiomqtt 2.x Client.__aexit__ behavior of re-raising the stored disconnect exception (if self._disconnected.done(): disconnect_exc = self._disconnected.exception(); ... raise disconnect_exc), which every 2.x release I know (2.0–2.4) has; I could not open the 2.5.1 source from…
| await client.subscribe( | ||
| incoming_topic, options=SubscribeOptions(qos=2, retainAsPublished=True, retainHandling=2) | ||
| ) |
There was a problem hiding this comment.
🔴 Callers whose peer topic the broker refuses to let them subscribe to get a transport that opens successfully and then never receives anything, instead of an error. At mqtt.py:84-86 the result of client.subscribe is discarded; aiomqtt only raises when the SUBSCRIBE packet cannot be sent and returns the SUBACK reason codes otherwise, so a broker answer such as 0x87 "Not authorized" (what Mosquitto sends for an ACL-denied topic) is treated as success. Fix: inspect the returned reason codes and raise aiomqtt.MqttError (or ValueError) when any code is a failure (>= 0x80), so the documented Raises: aiomqtt.MqttError: If subscription ... fails at mqtt.py:50 holds for broker-side refusals too. …
Extended reasoning...
…The README already flags the analogous publish-side gap as open; this one is fixable inside the adapter.
aiomqtt.Client.subscribe raises MqttCodeError only if paho cannot queue the SUBSCRIBE; it then awaits the SUBACK and returns its granted-QoS/reason-code list without checking it. mqtt.py:84 awaits that call and ignores the return value. With the shipped fixture (brokers/mosquitto.acl), a client whose ACL omits a read grant, or any deployment topic typo the ACL does not cover, gets SUBACK 0x87 from Mosquitto (MQTT 5) or 0x80 (3.1.1). mqtt_transport then yields the streams as if subscribed. read_messages waits on client.messages forever; the peer's replies are never delivered by the broker. On the client side, Client.aenter sends initialize (or the first call in 2026-07-28 mode) and waits read_timeout_seconds; demo_mqtt_disconnect.py passes read_timeout_seconds=None, so such a call hangs indefinitely. On the server side, runtime.connect returns normally and the connection idles with no log. The docstring at mqtt.py:50 promises MqttError when subscription fails, so callers wrap…
Verification: normal — triggers when the broker answers the SUBSCRIBE with a non-success SUBACK reason code (e.g. 0x87 Not authorized from a broker/plugin that enforces subscribe-time authorization, or 0x80 on 3.1.1). Mechanism: /home/claude/python-sdk/examples/transports/mcp_transport_examples/mqtt.py:84-86 does await client.subscribe(incoming_topic, options=SubscribeOptions(qos=2, ...)) and discards the…
| if message.retain or message.qos != 2 or len(message.payload) > max_message_size: | ||
| await send.send(ValueError("Rejected retained, non-QoS-2, or oversized MQTT message")) | ||
| continue | ||
| if not message.payload: | ||
| break | ||
| try: | ||
| decoded = jsonrpc_message_adapter.validate_json(message.payload, by_name=False) |
There was a problem hiding this comment.
🔴 If aiomqtt annotates Message.payload with its PayloadType union (str | bytes | bytearray | int | float | None, as aiomqtt 2.x documents), the new 'Check adapter types' CI step fails on this file. mqtt.py:69 passes message.payload to len() and mqtt.py:75 passes it to validate_json, and pyright strict rejects the int, float and None members; nothing narrows the type first. Fix: narrow once per message before use, e.g. bind payload = message.payload, send a ValueError item and continue unless it is bytes or bytearray, then use payload for the size check, the empty-payload close check and validate_json, which keeps paho's bytes payloads working.
Extended reasoning...
aiomqtt is not installed in this checkout, so I could not open aiomqtt/message.py; the condition is that Message.init takes payload: PayloadType and stores it unnarrowed, which is what aiomqtt 2.x's Message docstring lists as payload (str | bytes | bytearray | int | float | None). shared.yml:140-141 runs pyright --project examples/transports on every push. examples/transports/pyproject.toml:50-51 sets typeCheckingMode = 'strict' and includes mcp_transport_examples. len() requires Sized, so mqtt.py:69 reports reportArgumentType for int, float and None. At mqtt.py:75 the preceding if not message.payload: break only removes None and falsy values, so int and float remain and validate_json's str | bytes | bytearray parameter is still rejected. The repo forbids new # type: ignore (AGENTS.md), so the job stays red until the payload is narrowed. This step runs after the live broker demos in the same job, so a green demo run does not show it passing.
Verification: normal — triggered whenever the transport-examples CI job reaches its Check adapter types step with the locked aiomqtt 2.5.1 installed, provided aiomqtt 2.5.1 keeps the 2.x annotation Message.__init__(..., payload: PayloadType, ...) / self.payload = payload with PayloadType = str | bytes | bytearray | int | float | None (the annotation every aiomqtt 2.x release I know of uses and that…
| with anyio.move_on_after(1, shield=True), suppress(aiomqtt.MqttError): | ||
| await self.client.publish(self.topic, b"", qos=2, retain=False, properties=self.properties) |
There was a problem hiding this comment.
🟡 (optional) Servers keep a dead connection open forever, with its subscription, dispatcher task and lifespan slot, whenever a peer closes cleanly over a slow link. aclose at mqtt.py:120-121 waits at most 1 s for the QoS-2 close publish, then gives up while suppressing the error. If the broker's PUBREC has not arrived in that second, paho never sends PUBREL before the caller's clean DISCONNECT, so the broker discards the close and a clean DISCONNECT also discards the Last Will. The peer therefore gets neither signal. Fix: do not abandon the close signal on a fixed 1 s cap; wait for PUBCOMP for at least the configured expiry (or a caller-set close timeout), and apply the same bound to the unsubscribe at mqtt.py:97-98.
Extended reasoning...
The finder argued only that a broker round trip over 1 s cannot be sized here. MQTT's population is exactly slow links: cellular, satellite and congested IoT backhauls where 1 s round trips are routine. Every clean close on such a link runs this path. Step by step: the SDK closes the write stream, _MQTTWriter.__aexit__ calls aclose(). move_on_after(1, shield=True) wraps client.publish(..., qos=2). aiomqtt writes PUBLISH and awaits its confirmation future, which is only resolved on PUBCOMP. At 1 s the scope cancels the waiter; the transport's finally runs the 1 s unsubscribe; then the caller exits the aiomqtt client, which sends DISCONNECT immediately. Mosquitto only queues a QoS-2 inbound message to subscribers when it receives PUBREL (handle__pubrel -> db__message_release_incoming), and paho only sends PUBREL after PUBREC arrives. With PUBREC late, no PUBREL is ever sent; the clean DISCONNECT with session expiry 0 deletes the session and the half-completed message. MQTT 5 requires the Will to be discarded on a normal DISCONNECT, and the demo's keepalive-based will detection…
Verification: normal — when a peer closes cleanly while the broker's PUBREC for the empty-payload close publish arrives after the caller's DISCONNECT has been written (a link/broker round trip above roughly 2 s, TCP retransmission stalls, or a broker that is slow to acknowledge), the other side never sees the close and holds the session, subscription and read task indefinitely. Mechanism verified in the diff…
| keepalive=15, | ||
| will=aiomqtt.Will(f"{topic}/{outgoing}", payload=b"", qos=2, retain=False), | ||
| max_queued_incoming_messages=256, | ||
| ) |
There was a problem hiding this comment.
🟡 (optional) Users of this adapter silently lose requests or responses during a burst, and the affected call hangs until its read timeout, or forever when read_timeout_seconds=None as in demo_mqtt_disconnect.py:34. The demo sets max_queued_incoming_messages=256 at demo_mqtt.py:29; aiomqtt discards new messages with only a warning once that queue is full, after paho has already completed the QoS-2 handshake with the broker, so the broker never retransmits. The read path at mqtt.py:67-71 drains one message per several event-loop turns while paho enqueues one per turn, so any sustained burst above a few hundred messages overflows. …
Extended reasoning...
…Fix: never drop acknowledged messages; apply backpressure (block paho's reader or bound with an unbounded queue plus a stream-level ValueError) or fail the connection when the queue fills so the peer sees CONNECTION_CLOSED instead of a hang.
The finder dismissed this because README calls it an unresolved blocker and the CI demo only makes two calls; a documented drop is still silent data loss for anyone who copies the reference example, which is the stated purpose of the package. Trace: paho's socket reader runs once per loop iteration and delivers one PUBLISH per PUBREL to aiomqtt's on_message, which does queue.put_nowait. On QueueFull aiomqtt logs a warning and returns; paho has already sent PUBREC and will send PUBCOMP, so the broker treats delivery as complete. The consumer side, mqtt.py:67-71, iterates client.messages: each item costs a create_task for queue.get, an asyncio.wait, a yield, then a zero-buffer send that must rendezvous with the dispatcher's receive, then _dispatch. That is roughly four to six loop turns per message versus one to two per incoming packet. Under a…
Verification: normal; acknowledged in diff: examples/transports/README.md:105 ("The example bounds aiomqtt's incoming queue at 256 messages, but aiomqtt can drop messages when it fills... unresolved reliability blockers") and the PR description lists "queue-overflow behavior" as an open merge gate — the note is accurate about the bound but does not resolve the loss. Trigger: any burst where more than 256…
| Raises: | ||
| ValueError: If the configuration is invalid or an outgoing message is too large. | ||
| aiomqtt.MqttError: If subscription or publication fails. | ||
| """ |
There was a problem hiding this comment.
🟡 nit (optional): Maintainers get a Raises: section that lists argument-validation errors, which AGENTS.md says not to document. The ValueError entry at mqtt.py:49 describes invalid configuration (same topics, wildcards, expiry/size bounds), and its second clause ("outgoing message is too large") is raised by _MQTTWriter.send at mqtt.py:114, not by entering mqtt_transport(). Fix: keep only exceptions a caller would catch from entering the context (aiomqtt.MqttError), and move the oversized-message note to the description of max_message_size or to the writer.
Extended reasoning...
AGENTS.md Code Quality says: document exceptions a caller would reasonably catch in a Raises: section, and do not list exceptions from argument validation or programmer error. mqtt.py:48-51 lists ValueError: If the configuration is invalid or an outgoing message is too large. The configuration checks are mqtt.py:52-58 (aiomqtt.Topic(...), equal-topic check, expiry/size bounds), all argument validation. The oversized case is raised in _MQTTWriter.send at mqtt.py:113-114 when the dispatcher writes, so a reader catching ValueError around async with mqtt_transport(...) entry would not see it there. Consequence is documentation accuracy only; no runtime effect.
Verification: nit. Triggering condition: any reader of the mqtt_transport() docstring. AGENTS.md:51-53 (imported by root CLAUDE.md) states: "When a public API raises exceptions a caller would reasonably catch, document them in a Raises: section. Don't list exceptions from argument validation or programmer error." The new docstring at examples/transports/mcp_transport_examples/mqtt.py:48-50 lists… | nit.…
1 participant
Review scope
Stacked on #3519 for the shared example package and CI setup. This diff contains only the MQTT 5 adapter, Mosquitto configuration, broker examples, configuration tests, and MQTT documentation; there are no gRPC implementation changes.
The existing review follow-ups are preserved, including Last Will setup and client-ID isolation. Negative publish acknowledgments, queue-overflow behavior, broker recording/coverage, and production authorization remain open merge gates.
Validation
Six MQTT configuration tests and pre-commit pass after the split; the dedicated CI job also runs the live broker programs.
AI Disclaimer
This PR was developed with the assistance of either Claude or Codex. I've reviewed and verified the changes.