From 15fe1c20d1f5dd7fc6148202ed7acd933587b383 Mon Sep 17 00:00:00 2001
From: adiel-sammak <71227923+aelsammak@users.noreply.github.com>
Date: Mon, 28 Sep 2026 12:02:10 -0400
Subject: [PATCH] EBP-3104: Add best-practice Python request-reply reference
 samples

- Add python-direct-requestor-blocking-sample.py (publish_await_response with a positive reply timeout and one retry on PubSubTimeoutError), python-direct-requestor-sample.py (non-blocking publish() returning a Future per request, then collected), and python-direct-replier-sample.py (receive_async with a RequestMessageHandler, a None-replier guard, and reply())
- Direct only: the Python API has no guaranteed request-reply, so there is no counterpart to the JCSMP guaranteed requestor and replier
- Follow the EBP-3103 conventions: shared connection-config helper, listeners registered before connect(), 3000 ms reconnect interval, SIGINT/SIGTERM shutdown, bounded terminate(), exit code 1 on failure and 0 on a shutdown signal
- Document that the request-reply builders have no back-pressure setting and that their from_properties() applies no property
- Rename the four EBP-3103 samples to drop the -async suffix: a -blocking suffix now marks only a variant that waits for the broker's answer, matching the SolaceSamples naming
- Correct a stale guaranteed publisher comment that said the direct publisher pauses during a reconnect
- Verified against solace-pubsubplus 1.11.0: pyflakes clean, a no-broker smoke test, and live round trips from both requestors to the replier
---
 ...e.py => python-direct-publisher-sample.py} |   2 +-
 .../python/python-direct-replier-sample.py    | 316 +++++++++++++++++
 ...python-direct-requestor-blocking-sample.py | 310 +++++++++++++++++
 .../python/python-direct-requestor-sample.py  | 324 ++++++++++++++++++
 ....py => python-direct-subscriber-sample.py} |   2 +-
 ... => python-guaranteed-publisher-sample.py} |  10 +-
 ...=> python-guaranteed-subscriber-sample.py} |   2 +-
 7 files changed, 958 insertions(+), 8 deletions(-)
 rename plugins/solace-messaging-skills/skills/solace-application-development/references/python/{python-direct-publisher-async-sample.py => python-direct-publisher-sample.py} (99%)
 create mode 100644 plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-replier-sample.py
 create mode 100644 plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-blocking-sample.py
 create mode 100644 plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-sample.py
 rename plugins/solace-messaging-skills/skills/solace-application-development/references/python/{python-direct-subscriber-async-sample.py => python-direct-subscriber-sample.py} (99%)
 rename plugins/solace-messaging-skills/skills/solace-application-development/references/python/{python-guaranteed-publisher-async-sample.py => python-guaranteed-publisher-sample.py} (97%)
 rename plugins/solace-messaging-skills/skills/solace-application-development/references/python/{python-guaranteed-subscriber-async-sample.py => python-guaranteed-subscriber-sample.py} (99%)

diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-sample.py
similarity index 99%
rename from plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-async-sample.py
rename to plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-sample.py
index 1ea90bd..8a5d262 100644
--- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-async-sample.py
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-publisher-sample.py
@@ -32,7 +32,7 @@
 
 Only practices documented in canonical Solace sources are encoded here.
 
-Copied into a generated project as direct_publisher_async.py, next to the shared helper
+Copied into a generated project as direct_publisher.py, next to the shared helper
 python-solace-connection-config.py copied as solace_connection_config.py.
 
 Any generated adaptation of this sample MUST begin with the exact line:
diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-replier-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-replier-sample.py
new file mode 100644
index 0000000..b5e1c11
--- /dev/null
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-replier-sample.py
@@ -0,0 +1,316 @@
+"""
+Derived from patterns/direct_replier.py
+  SolaceSamples/solace-samples-python
+  https://github.com/SolaceSamples/solace-samples-python
+
+Elevated to documented best practices per the Python API developer guide:
+  Messaging Service:
+    https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md
+  Request-Reply Messaging:
+    https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Request-Reply.md
+  the Python API reference (units, defaults, listener contracts):
+    https://docs.solace.com/API-Developer-Online-Ref-Documentation/python/index.html
+  and the C API Best Practices (the Python API wraps the C API, so its callback,
+  reconnect, acknowledgement, and time-to-live practices apply):
+    https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md
+
+Reference sample. Wired for a basic direct request-reply journey on the replier
+side: basic-auth connect with a reconnection retry strategy, a
+RequestReplyMessageReceiver built with the request topic subscription (no queue, no
+provisioning), an async RequestMessageHandler that guards on a replier (a request
+without a reply-to cannot be answered) and answers each request with
+replier.reply(), which routes to the request's reply-to and copies the correlation,
+reconnection and service interruption listeners registered BEFORE connect(), and a
+graceful SIGINT/SIGTERM shutdown. Request-reply in the Python API is direct only, so
+it is at-most-once on both legs. Answers both requestor samples,
+python-direct-requestor-blocking-sample.py and python-direct-requestor-sample.py.
+Structured as module-level functions, setup_solace(...), connect_solace(...),
+await_requests(...), and teardown_solace(...), that share one ReplierState;
+setup_solace() creates the service and the replier before any connection exists, and
+main() runs teardown from its finally on every exit path after that.
+
+Only practices documented in canonical Solace sources are encoded here.
+
+Copied into a generated project as direct_replier.py, next to the shared helper
+python-solace-connection-config.py copied as solace_connection_config.py.
+
+Any generated adaptation of this sample MUST begin with the exact line:
+  AI-assisted code. Review before production use.
+(This reference sample itself carries no such header by design.)
+"""
+
+import logging
+import signal
+import sys
+import threading
+from dataclasses import dataclass
+
+from solace.messaging.config.retry_strategy import RetryStrategy
+from solace.messaging.config.solace_properties import message_properties
+from solace.messaging.errors.pubsubplus_client_error import IncompleteMessageDeliveryError, PubSubPlusClientError
+from solace.messaging.messaging_service import (
+    MessagingService,
+    ReconnectionAttemptListener,
+    ReconnectionListener,
+    ServiceEvent,
+    ServiceInterruptionListener,
+)
+from solace.messaging.receiver.inbound_message import InboundMessage
+from solace.messaging.receiver.request_reply_message_receiver import (
+    Replier,
+    RequestMessageHandler,
+    RequestReplyMessageReceiver,
+)
+from solace.messaging.resources.topic_subscription import TopicSubscription
+
+from solace_connection_config import load_service_properties
+
+APP_NAME = "DirectReplier"
+# request topic to subscribe to; the requestor samples send to this topic
+REQUEST_TOPIC = "solace/samples/python/direct/request"
+API = "Python"
+# reconnect budget: 20 attempts 3 s apart, a minute of reconnect attempts, a reasonable
+# default when the design says nothing about high availability. The interval is in
+# MILLISECONDS per the RetryStrategy API reference.
+RECONNECT_RETRIES = 20
+RECONNECT_RETRY_INTERVAL_MS = 3000
+# For HA failover resilience, use the C API Best Practices values
+# (https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md):
+#   HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)),
+#     with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)),
+#     properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5
+# terminate(grace_period): how long to wait for the handler to answer the requests the API
+# has already received before the replier stops; terminate() raises
+# IncompleteMessageDeliveryError when requests remain after that. A bounded grace period
+# keeps shutdown time predictable, for example within the stop timeout of a process
+# supervisor, so the disconnect still runs. The API default is 600000 ms (10 minutes).
+# Adjust the value to suit the application: longer lets the handler answer more of the
+# received requests, shorter exits sooner.
+TERMINATE_GRACE_PERIOD_MS = 10_000
+
+# API events go through standard logging; the trace() narration below is a separate channel
+logger = logging.getLogger(APP_NAME)
+
+
+@dataclass
+class ReplierState:
+    """What the main loop, the signal handler, and the API listener threads share: the
+    service, the replier, the shutdown event, and the counters. The service and the
+    replier are required, so every ReplierState holds both (built, though not necessarily
+    connected or started). Use this type of app to answer direct (at-most-once) requests
+    on a topic."""
+    messaging_service: MessagingService
+    receiver: RequestReplyMessageReceiver
+    shutdown: threading.Event
+    connect_attempted: bool = False
+    exit_code: int = 0  # set to 1 on a failure exit (service interruption or a failed terminate())
+    msg_recv_counter: int = 0  # num requests received
+
+
+def main(args: list[str]) -> int:
+    trace(f"{API} {APP_NAME} initializing...")
+    shutdown = threading.Event()
+
+    def on_shutdown_signal(signum: int, _frame) -> None:
+        trace(f"Shutdown signal received ({signal.Signals(signum).name}), stopping replier...")
+        shutdown.set()
+
+    # graceful shutdown: SIGINT (Ctrl-C) or SIGTERM sets the shutdown event, the main
+    # loop exits, and teardown_solace() in the finally below stops the replier and
+    # disconnects. Python runs signal handlers on the main thread, so the handler only
+    # flags; the cleanup runs on the main thread's normal exit path. Registered before
+    # connect, so a SIGTERM during the blocking connect() still reaches teardown.
+    signal.signal(signal.SIGINT, on_shutdown_signal)
+    signal.signal(signal.SIGTERM, on_shutdown_signal)
+    # basic username/password connection details, built by the shared connection-config
+    # helper: read from a config.json in the working directory if present, else from the
+    # command line (<host:port> <message-vpn> <client-username> [password]). Turning the
+    # application's input into Solace properties is the application's job, so it stays out
+    # of setup_solace(), which works from the properties alone.
+    properties = load_service_properties(args, APP_NAME)
+    # setup_solace() raises before any connection exists, so there is nothing to tear down
+    # if it fails
+    state = setup_solace(properties, shutdown)
+    try:
+        connect_solace(state)
+        await_requests(state)
+    finally:
+        teardown_solace(state)
+        trace("Main thread quitting.")
+    return state.exit_code  # non-zero after a failure, so scripts and supervisors see it
+
+
+def setup_solace(properties: dict, shutdown: threading.Event) -> ReplierState:
+    # build() creates the native session and resolves the host, so an unresolvable host
+    # fails here, before connect()
+    messaging_service = (
+        MessagingService.builder()
+        # the builder merges each call in order and a later call wins, so the reconnection
+        # strategy comes first: it is the default, and reconnection keys in the properties
+        # (for example from config.json) override it
+        .with_reconnection_retry_strategy(
+            RetryStrategy.parametrized_retry(RECONNECT_RETRIES, RECONNECT_RETRY_INTERVAL_MS))
+        .from_properties(properties)
+        .build()
+    )
+    # request-reply receiver: the request topic subscription goes to build() (no queue, no
+    # provisioning). start() applies the subscription and blocks until the broker confirms
+    # it, so the route is live before this replier reports ready. Its builder has no
+    # back-pressure setting (unlike the direct receiver), and its from_properties() applies
+    # no property, so there is nothing more to configure. To spread requests across several
+    # replier instances, pass a share name: build(subscription, ShareName.of("name")), and
+    # each request then goes to one member of the group. Building the receiver needs only
+    # the built service; start() is what requires the connection.
+    receiver = (
+        messaging_service.request_reply()
+        .create_request_reply_message_receiver_builder()
+        .build(TopicSubscription.of(REQUEST_TOPIC))
+    )
+    state = ReplierState(messaging_service=messaging_service, receiver=receiver, shutdown=shutdown)
+    # best practice: register the service event listeners BEFORE connect(), so no
+    # reconnection or interruption event raised during or right after the connect is lost,
+    # and handle each event appropriately rather than only logging it
+    service_events = ServiceEventHandler(state)
+    messaging_service.add_reconnection_attempt_listener(service_events)
+    messaging_service.add_reconnection_listener(service_events)
+    messaging_service.add_service_interruption_listener(service_events)
+    return state
+
+
+def connect_solace(state: ReplierState) -> None:
+    state.connect_attempted = True
+    state.messaging_service.connect()  # blocking connect
+    # set the request handler BEFORE start(), so the handler is in place when delivery
+    # begins and received requests do not wait in the API's buffer for it. receive_async()
+    # needs a connected service, so it runs here after connect(), not in setup_solace().
+    # see bottom of file for RequestHandler, which answers the requests from the topic
+    state.receiver.receive_async(RequestHandler(state))
+    trace(f"Adding direct request-topic subscription '{REQUEST_TOPIC}'.")
+    state.receiver.start()
+    trace(f"{APP_NAME} subscribed and ready to reply. Press Ctrl-C to quit.")
+
+
+def await_requests(state: ReplierState) -> None:
+    # async request receive working now, so time to wait until done...
+    while not state.shutdown.wait(1.0):  # wait 1 second; the wait returns True once shutdown is requested
+        trace(f"{API} {APP_NAME} Requests/s: {state.msg_recv_counter:,}")  # simple way of calculating request rates
+        state.msg_recv_counter = 0
+
+
+def teardown_solace(state: ReplierState) -> None:
+    # Application cleanup belongs here: once setup_solace() returns, teardown_solace() runs
+    # in main's finally on EVERY exit path (normal quit, SIGINT/SIGTERM, service
+    # interruption, or an exception), so Solace teardown and application-side cleanup are
+    # never skipped.
+    state.shutdown.set()
+    if state.receiver.is_running():
+        # request-reply is at-most-once: there are no acknowledgements to drain before exit,
+        # so terminate() stops delivery and gives the handler up to the grace period to answer
+        # the requests the API has already received. It raises IncompleteMessageDeliveryError
+        # when requests remain after that; catch it so the disconnect below still runs
+        try:
+            state.receiver.terminate(TERMINATE_GRACE_PERIOD_MS)
+        except IncompleteMessageDeliveryError as error:
+            logger.error("Replier stopped with requests still unanswered after %d ms: %s",
+                         TERMINATE_GRACE_PERIOD_MS, error)
+            state.exit_code = 1
+        except PubSubPlusClientError as error:
+            logger.error("Replier terminate() failed: %s", error)
+            state.exit_code = 1
+    if state.connect_attempted:
+        # disconnect() raises IllegalStateError on a service that never attempted to connect,
+        # hence the flag; on a service that is already down it returns quietly
+        state.messaging_service.disconnect()  # will also release the replier
+
+
+def trace(message: str) -> None:
+    """Demo narration sink: every status line in this sample funnels through this one
+    function. An application replaces this single body to route narration to its
+    logger or reporting system. The logger calls for API events (service, reply)
+    are a separate channel and stay as they are."""
+    print(message, flush=True)
+
+
+############################################################################
+
+
+class ServiceEventHandler(ReconnectionAttemptListener, ReconnectionListener, ServiceInterruptionListener):
+    """The three service event listeners in one class. Callbacks run on an API thread:
+    return promptly and do not block in them, since events and delivery stall while a
+    callback runs (C API Best Practices)."""
+
+    def __init__(self, state: ReplierState) -> None:
+        self.state = state
+
+    def on_reconnecting(self, event: ServiceEvent) -> None:
+        # connection lost, automatic reconnect attempt in progress: request delivery is
+        # paused, and the broker does not store direct requests for this replier while it
+        # is disconnected (at-most-once)
+        logger.warning("Service reconnecting; request delivery is paused until re-established: %s (%s)",
+                       event.get_message(), event.get_cause())
+
+    def on_reconnected(self, event: ServiceEvent) -> None:
+        # automatic reconnect succeeded; the API re-applies the request topic subscription and
+        # delivery resumes (unlike JCSMP, no reapply-subscriptions property is needed)
+        logger.info("Service reconnected to %s; the subscription is restored and replies resume",
+                    event.get_broker_uri())
+
+    def on_service_interrupted(self, event: ServiceEvent) -> None:
+        # the connection went down and cannot be restored (reconnect attempts exhausted)
+        logger.error("Service interrupted and cannot be restored, quitting: %s (%s)",
+                     event.get_message(), event.get_cause())
+        # Application cleanup signal: the service will not recover. Trigger application-side
+        # cleanup from here. This sample sets shutdown, so the main loop exits and
+        # teardown_solace() runs in main's finally.
+        self.state.exit_code = 1
+        self.state.shutdown.set()
+
+
+class RequestHandler(RequestMessageHandler):
+    """Very simple class, used for answering the direct requests from the topic. on_message
+    runs on an API-owned receiver thread, not the main thread: return promptly and do not
+    block in it (hand heavy work to the application's own thread or queue), since delivery
+    stalls while it runs (C API Best Practices)."""
+
+    def __init__(self, state: ReplierState) -> None:
+        self.state = state
+        # one OutboundMessageBuilder for every reply; the per-reply difference goes in
+        # build()'s additional_message_properties. Replier.reply() takes an OutboundMessage,
+        # not a bare payload, so the builder is required here
+        self.message_builder = state.messaging_service.message_builder()
+
+    def on_message(self, message: InboundMessage, replier: Replier) -> None:
+        self.state.msg_recv_counter += 1
+        # guard on the replier: the API passes None when the request has no reply-to
+        # destination, and such a message cannot be answered. The requestor samples set the
+        # reply-to automatically; a message arriving here without one is not a request.
+        if replier is None:
+            logger.warning("Received a message on the request topic with no reply-to; ignoring.")
+            return
+        try:
+            # the requestor samples send a bytearray payload (the binary attachment), read back
+            # with get_payload_as_bytes(); a str payload arrives via get_payload_as_string()
+            # instead. best practice: handle an unexpected message format without raising (C
+            # API Best Practices); errors="replace" keeps a non-UTF-8 payload from raising
+            payload = message.get_payload_as_bytes()
+            text = payload.decode("utf-8", errors="replace") if payload is not None else message.get_payload_as_string()
+            # echo the request's application message id onto the reply for traceability; the
+            # API copies the correlation itself, so the requestor matches the reply without it
+            msg_id = message.get_application_message_id()
+            reply_properties = {message_properties.APPLICATION_MESSAGE_ID: msg_id} if msg_id is not None else None
+            reply = self.message_builder.build(
+                bytearray(f"Reply from {APP_NAME} to: {text}".encode("utf-8")),
+                additional_message_properties=reply_properties)
+            # reply() routes to the request's reply-to and copies the correlation id, so the
+            # requestor's publish_await_response() or publish() Future receives it
+            replier.reply(reply)
+        except PubSubPlusClientError as error:
+            # the reply could not be sent (for example the service is down); the requestor
+            # times out. Without this catch, the API logs the exception as one warning.
+            logger.warning("Could not send the reply for message id %s: %s",
+                           message.get_application_message_id(), error)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    sys.exit(main(sys.argv[1:]))
diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-blocking-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-blocking-sample.py
new file mode 100644
index 0000000..7ad5d69
--- /dev/null
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-blocking-sample.py
@@ -0,0 +1,310 @@
+"""
+Derived from patterns/direct_requestor_blocking.py
+  SolaceSamples/solace-samples-python
+  https://github.com/SolaceSamples/solace-samples-python
+
+Elevated to documented best practices per the Python API developer guide:
+  Messaging Service:
+    https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md
+  Request-Reply Messaging:
+    https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Request-Reply.md
+  the Python API reference (units, defaults, listener contracts):
+    https://docs.solace.com/API-Developer-Online-Ref-Documentation/python/index.html
+  and the C API Best Practices (the Python API wraps the C API, so its callback,
+  reconnect, acknowledgement, and time-to-live practices apply):
+    https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md
+
+Reference sample. Wired for a basic direct request-reply journey on the requestor
+side, BLOCKING variant: basic-auth connect with a reconnection retry strategy, a
+RequestReplyMessagePublisher, one blocking publish_await_response() with a POSITIVE
+reply timeout, a PubSubTimeoutError catch with a single retry to absorb a cold-start
+race, reconnection and service interruption listeners registered BEFORE connect(),
+and a graceful SIGINT/SIGTERM shutdown. The blocking request IS the synchronization
+point: the requestor sends one request, waits for the correlated reply, and exits
+(0 on a reply or a shutdown signal, 1 without a reply). Request-reply in the Python API is direct only, so
+it is at-most-once on both legs. The non-blocking variant is
+python-direct-requestor-sample.py. Structured as module-level functions,
+setup_solace(...), connect_solace(...), send_request(...), and teardown_solace(...),
+that share one RequestorState; setup_solace() creates the service and the requestor
+before any connection exists, and main() runs teardown from its finally on every
+exit path after that.
+
+Only practices documented in canonical Solace sources are encoded here.
+
+Copied into a generated project as direct_requestor_blocking.py, next to the shared
+helper python-solace-connection-config.py copied as solace_connection_config.py.
+
+Any generated adaptation of this sample MUST begin with the exact line:
+  AI-assisted code. Review before production use.
+(This reference sample itself carries no such header by design.)
+"""
+
+import logging
+import signal
+import sys
+import threading
+import uuid
+from dataclasses import dataclass
+
+from solace.messaging.config.retry_strategy import RetryStrategy
+from solace.messaging.config.solace_properties import message_properties
+from solace.messaging.errors.pubsubplus_client_error import (
+    IncompleteMessageDeliveryError,
+    PubSubPlusClientError,
+    PubSubTimeoutError,
+)
+from solace.messaging.messaging_service import (
+    MessagingService,
+    ReconnectionAttemptListener,
+    ReconnectionListener,
+    ServiceEvent,
+    ServiceInterruptionListener,
+)
+from solace.messaging.publisher.request_reply_message_publisher import RequestReplyMessagePublisher
+from solace.messaging.resources.topic import Topic
+
+from solace_connection_config import load_service_properties
+
+APP_NAME = "DirectRequestorBlocking"
+# request topic the replier subscribes to; matches the direct replier sample
+REQUEST_TOPIC = "solace/samples/python/direct/request"
+API = "Python"
+# POSITIVE reply timeout in MILLISECONDS: publish_await_response() blocks up to this long
+# for the correlated reply, then raises PubSubTimeoutError. The API has no default and
+# requires a positive value.
+REQUEST_TIMEOUT_MS = 3000
+# one retry after a timeout absorbs a cold-start race: the replier may not yet have its
+# subscription in place when the first request goes out
+REQUEST_ATTEMPTS = 2
+# reconnect budget: 20 attempts 3 s apart, a minute of reconnect attempts, a reasonable
+# default when the design says nothing about high availability. The interval is in
+# MILLISECONDS per the RetryStrategy API reference.
+RECONNECT_RETRIES = 20
+RECONNECT_RETRY_INTERVAL_MS = 3000
+# For HA failover resilience, use the C API Best Practices values
+# (https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md):
+#   HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)),
+#     with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)),
+#     properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5
+# terminate(grace_period): how long to wait for a request still waiting on its reply before
+# the requestor stops; terminate() raises IncompleteMessageDeliveryError when a request is
+# still outstanding after that. A bounded grace period keeps shutdown time predictable, for
+# example within the stop timeout of a process supervisor, so the disconnect still runs.
+# The API default is 600000 ms (10 minutes). Adjust the value to suit the application:
+# longer gives an outstanding request more time to get its reply, shorter exits sooner.
+TERMINATE_GRACE_PERIOD_MS = 10_000
+
+# API events go through standard logging; the trace() narration below is a separate channel
+logger = logging.getLogger(APP_NAME)
+
+
+@dataclass
+class RequestorState:
+    """What the main thread, the signal handler, and the API listener threads share: the
+    service, the requestor, and the shutdown event. The service and the requestor are
+    required, so every RequestorState holds both (built, though not necessarily connected
+    or started)."""
+    messaging_service: MessagingService
+    requestor: RequestReplyMessagePublisher
+    shutdown: threading.Event
+    connect_attempted: bool = False
+    exit_code: int = 0  # set to 1 on a failure exit (no reply, service interruption, or a failed terminate())
+
+
+def main(args: list[str]) -> int:
+    trace(f"{API} {APP_NAME} initializing...")
+    shutdown = threading.Event()
+
+    def on_shutdown_signal(signum: int, _frame) -> None:
+        trace(f"Shutdown signal received ({signal.Signals(signum).name}), stopping requestor...")
+        shutdown.set()
+
+    # graceful shutdown: SIGINT (Ctrl-C) or SIGTERM sets the shutdown event, no further
+    # request goes out, and teardown_solace() in the finally below stops the requestor and
+    # disconnects. Python runs signal handlers on the main thread, so the handler only
+    # flags; the cleanup runs on the main thread's normal exit path. Registered before
+    # connect, so a SIGTERM during the blocking connect() still reaches teardown.
+    signal.signal(signal.SIGINT, on_shutdown_signal)
+    signal.signal(signal.SIGTERM, on_shutdown_signal)
+    # basic username/password connection details, built by the shared connection-config
+    # helper: read from a config.json in the working directory if present, else from the
+    # command line (<host:port> <message-vpn> <client-username> [password]). Turning the
+    # application's input into Solace properties is the application's job, so it stays out
+    # of setup_solace(), which works from the properties alone.
+    properties = load_service_properties(args, APP_NAME)
+    # setup_solace() raises before any connection exists, so there is nothing to tear down
+    # if it fails
+    state = setup_solace(properties, shutdown)
+    try:
+        connect_solace(state)
+        send_request(state)
+    finally:
+        teardown_solace(state)
+        trace("Main thread quitting.")
+    return state.exit_code  # non-zero after a failure, so scripts and supervisors see it
+
+
+def setup_solace(properties: dict, shutdown: threading.Event) -> RequestorState:
+    # build() creates the native session and resolves the host, so an unresolvable host
+    # fails here, before connect()
+    messaging_service = (
+        MessagingService.builder()
+        # the builder merges each call in order and a later call wins, so the reconnection
+        # strategy comes first: it is the default, and reconnection keys in the properties
+        # (for example from config.json) override it
+        .with_reconnection_retry_strategy(
+            RetryStrategy.parametrized_retry(RECONNECT_RETRIES, RECONNECT_RETRY_INTERVAL_MS))
+        .from_properties(properties)
+        .build()
+    )
+    # request-reply publisher: the API sets a unique reply-to topic on each request and
+    # correlates the reply, so the requestor needs no reply subscription and no correlation
+    # id of its own. Its builder has no back-pressure setting (unlike the direct publisher),
+    # and its from_properties() applies no property, so there is nothing more to configure.
+    # Building the requestor needs only the built service; start() is what requires the
+    # connection.
+    requestor = (
+        messaging_service.request_reply()
+        .create_request_reply_message_publisher_builder()
+        .build()
+    )
+    state = RequestorState(messaging_service=messaging_service, requestor=requestor, shutdown=shutdown)
+    # best practice: register the service event listeners BEFORE connect(), so no
+    # reconnection or interruption event raised during or right after the connect is lost,
+    # and handle each event appropriately rather than only logging it
+    service_events = ServiceEventHandler(state)
+    messaging_service.add_reconnection_attempt_listener(service_events)
+    messaging_service.add_reconnection_listener(service_events)
+    messaging_service.add_service_interruption_listener(service_events)
+    return state
+
+
+def connect_solace(state: RequestorState) -> None:
+    state.connect_attempted = True
+    state.messaging_service.connect()  # blocking connect
+    state.requestor.start()  # raises IllegalStateError unless the service is connected
+
+
+def send_request(state: RequestorState) -> None:
+    # the request is an OutboundMessage: publish_await_response() takes no bare payload, so
+    # the message builder is required here. A bytearray payload travels as the binary
+    # attachment; the replier reads it back with get_payload_as_bytes()
+    request = state.messaging_service.message_builder().build(
+        bytearray(f"Request from {APP_NAME}".encode("utf-8")))
+    # an application message id lets the replier echo it onto the reply for traceability;
+    # the API does the request-reply correlation itself and does not need it
+    msg_id = str(uuid.uuid4())
+    request_topic = Topic.of(REQUEST_TOPIC)
+    trace(f"{APP_NAME} sending a direct request to topic '{REQUEST_TOPIC}' (timeout {REQUEST_TIMEOUT_MS} ms).")
+    reply = None
+    for attempt in range(1, REQUEST_ATTEMPTS + 1):
+        if state.shutdown.is_set():
+            break  # a shutdown signal or a service interruption: send no further request
+        try:
+            # blocks until the correlated reply arrives or REQUEST_TIMEOUT_MS passes. A
+            # shutdown signal that arrives during the wait takes effect once the call
+            # returns, within REQUEST_TIMEOUT_MS.
+            reply = state.requestor.publish_await_response(
+                request_message=request,
+                request_destination=request_topic,
+                reply_timeout=REQUEST_TIMEOUT_MS,
+                additional_message_properties={message_properties.APPLICATION_MESSAGE_ID: msg_id},
+            )
+            break
+        except PubSubTimeoutError:
+            if attempt < REQUEST_ATTEMPTS:
+                logger.warning("Request timed out after %d ms; retrying once.", REQUEST_TIMEOUT_MS)
+            else:
+                logger.warning("Request timed out after %d ms on the retry; giving up.", REQUEST_TIMEOUT_MS)
+        except PubSubPlusClientError as error:
+            # the request could not be sent (for example the service is down); retrying would
+            # not help
+            logger.error("Request failed, quitting: %s", error)
+            break
+
+    if reply is None:
+        if state.shutdown.is_set():
+            # a shutdown signal is a clean stop (exit 0); a service interruption has already
+            # set exit_code
+            trace(f"{API} {APP_NAME} stopped before a reply arrived.")
+        else:
+            trace(f"{API} {APP_NAME} received no reply.")
+            state.exit_code = 1
+        return
+    # the replier sample sends a bytearray payload (the binary attachment), read back with
+    # get_payload_as_bytes(); a str payload arrives via get_payload_as_string() instead.
+    # errors="replace" keeps a non-UTF-8 payload from raising
+    payload = reply.get_payload_as_bytes()
+    text = payload.decode("utf-8", errors="replace") if payload is not None else reply.get_payload_as_string()
+    trace(f"{API} {APP_NAME} received a correlated reply (message id {reply.get_application_message_id()}): {text}")
+
+
+def teardown_solace(state: RequestorState) -> None:
+    # Application cleanup belongs here: once setup_solace() returns, teardown_solace() runs
+    # in main's finally on EVERY exit path (a reply, no reply, SIGINT/SIGTERM, service
+    # interruption, or an exception), so Solace teardown and application-side cleanup are
+    # never skipped.
+    state.shutdown.set()
+    if state.requestor.is_running():
+        # terminate() gives a request still waiting on its reply up to the grace period. It
+        # raises IncompleteMessageDeliveryError when a request is still outstanding after
+        # that; catch it so the disconnect below still runs
+        try:
+            state.requestor.terminate(TERMINATE_GRACE_PERIOD_MS)
+        except IncompleteMessageDeliveryError as error:
+            logger.error("Requestor stopped with a request still outstanding after %d ms: %s",
+                         TERMINATE_GRACE_PERIOD_MS, error)
+            state.exit_code = 1
+        except PubSubPlusClientError as error:
+            logger.error("Requestor terminate() failed: %s", error)
+            state.exit_code = 1
+    if state.connect_attempted:
+        # disconnect() raises IllegalStateError on a service that never attempted to connect,
+        # hence the flag; on a service that is already down it returns quietly
+        state.messaging_service.disconnect()
+
+
+def trace(message: str) -> None:
+    """Demo narration sink: every status line in this sample funnels through this one
+    function. An application replaces this single body to route narration to its
+    logger or reporting system. The logger calls for API events (service, request)
+    are a separate channel and stay as they are."""
+    print(message, flush=True)
+
+
+############################################################################
+
+
+class ServiceEventHandler(ReconnectionAttemptListener, ReconnectionListener, ServiceInterruptionListener):
+    """The three service event listeners in one class. Callbacks run on an API thread:
+    return promptly and do not block in them, since events and delivery stall while a
+    callback runs (C API Best Practices)."""
+
+    def __init__(self, state: RequestorState) -> None:
+        self.state = state
+
+    def on_reconnecting(self, event: ServiceEvent) -> None:
+        # connection lost, automatic reconnect attempt in progress. Request-reply is direct
+        # (at-most-once): a request or reply in flight during the outage can be lost, and the
+        # request then times out
+        logger.warning("Service reconnecting; a request in flight can time out: %s (%s)",
+                       event.get_message(), event.get_cause())
+
+    def on_reconnected(self, event: ServiceEvent) -> None:
+        # automatic reconnect succeeded: requests can go out again
+        logger.info("Service reconnected to %s", event.get_broker_uri())
+
+    def on_service_interrupted(self, event: ServiceEvent) -> None:
+        # the connection went down and cannot be restored (reconnect attempts exhausted)
+        logger.error("Service interrupted and cannot be restored, quitting: %s (%s)",
+                     event.get_message(), event.get_cause())
+        # Application cleanup signal: the service will not recover. Trigger application-side
+        # cleanup from here. This sample sets shutdown, so no further request goes out, the
+        # request in flight fails or times out, and teardown_solace() runs in main's finally.
+        self.state.exit_code = 1
+        self.state.shutdown.set()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    sys.exit(main(sys.argv[1:]))
diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-sample.py
new file mode 100644
index 0000000..a9807a7
--- /dev/null
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-requestor-sample.py
@@ -0,0 +1,324 @@
+"""
+Derived from patterns/direct_requestor.py
+  SolaceSamples/solace-samples-python
+  https://github.com/SolaceSamples/solace-samples-python
+
+Elevated to documented best practices per the Python API developer guide:
+  Messaging Service:
+    https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Messaging-Service.md
+  Request-Reply Messaging:
+    https://docs.solace.com/API/API-Developer-Guide-Python/Python-API-Request-Reply.md
+  the Python API reference (units, defaults, listener contracts):
+    https://docs.solace.com/API-Developer-Online-Ref-Documentation/python/index.html
+  and the C API Best Practices (the Python API wraps the C API, so its callback,
+  reconnect, acknowledgement, and time-to-live practices apply):
+    https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md
+
+Reference sample. Wired for a basic direct request-reply journey on the requestor
+side, NON-BLOCKING variant: basic-auth connect with a reconnection retry strategy, a
+RequestReplyMessagePublisher, several non-blocking publish() calls with a POSITIVE
+reply timeout that each return a concurrent.futures.Future at once, so the requests
+are in flight together, then each Future collected for its correlated reply or its
+PubSubTimeoutError, reconnection and service interruption listeners registered
+BEFORE connect(), and a graceful SIGINT/SIGTERM shutdown. The requestor sends its
+requests, collects every reply, and exits (0 when every request sent got a reply, 1
+otherwise). Request-reply in the Python API is direct only, so it is at-most-once on
+both legs. The blocking variant, one request at a time, is
+python-direct-requestor-blocking-sample.py. Structured as module-level functions,
+setup_solace(...), connect_solace(...), send_requests(...), and teardown_solace(...),
+that share one RequestorState; setup_solace() creates the service and the requestor
+before any connection exists, and main() runs teardown from its finally on every
+exit path after that.
+
+Only practices documented in canonical Solace sources are encoded here.
+
+Copied into a generated project as direct_requestor.py, next to the shared helper
+python-solace-connection-config.py copied as solace_connection_config.py.
+
+Any generated adaptation of this sample MUST begin with the exact line:
+  AI-assisted code. Review before production use.
+(This reference sample itself carries no such header by design.)
+"""
+
+import concurrent.futures
+import logging
+import signal
+import sys
+import threading
+import uuid
+from dataclasses import dataclass
+
+from solace.messaging.config.retry_strategy import RetryStrategy
+from solace.messaging.config.solace_properties import message_properties
+from solace.messaging.errors.pubsubplus_client_error import (
+    IncompleteMessageDeliveryError,
+    PubSubPlusClientError,
+    PubSubTimeoutError,
+)
+from solace.messaging.messaging_service import (
+    MessagingService,
+    ReconnectionAttemptListener,
+    ReconnectionListener,
+    ServiceEvent,
+    ServiceInterruptionListener,
+)
+from solace.messaging.publisher.request_reply_message_publisher import RequestReplyMessagePublisher
+from solace.messaging.resources.topic import Topic
+
+from solace_connection_config import load_service_properties
+
+APP_NAME = "DirectRequestor"
+# request topic the replier subscribes to; matches the direct replier sample
+REQUEST_TOPIC = "solace/samples/python/direct/request"
+API = "Python"
+# how many requests go out before the first reply is collected
+REQUEST_COUNT = 5
+# POSITIVE reply timeout in MILLISECONDS, per request: each Future completes with the
+# correlated reply, or with PubSubTimeoutError once this long passes without one. The API
+# has no default and requires a positive value.
+REQUEST_TIMEOUT_MS = 3000
+# reconnect budget: 20 attempts 3 s apart, a minute of reconnect attempts, a reasonable
+# default when the design says nothing about high availability. The interval is in
+# MILLISECONDS per the RetryStrategy API reference.
+RECONNECT_RETRIES = 20
+RECONNECT_RETRY_INTERVAL_MS = 3000
+# For HA failover resilience, use the C API Best Practices values
+# (https://docs.solace.com/API/API-Developer-Guide/C-API-Best-Practices.md):
+#   HA failover: with_connection_retry_strategy(RetryStrategy.parametrized_retry(1, 3000)),
+#     with_reconnection_retry_strategy(RetryStrategy.parametrized_retry(20, 3000)),
+#     properties[transport_layer_properties.CONNECTION_RETRIES_PER_HOST] = 5
+# terminate(grace_period): how long to wait for requests still waiting on their replies
+# before the requestor stops; terminate() raises IncompleteMessageDeliveryError when
+# requests are still outstanding after that. A bounded grace period keeps shutdown time
+# predictable, for example within the stop timeout of a process supervisor, so the
+# disconnect still runs. The API default is 600000 ms (10 minutes). Adjust the value to
+# suit the application: longer gives outstanding requests more time to get their replies,
+# shorter exits sooner.
+TERMINATE_GRACE_PERIOD_MS = 10_000
+
+# API events go through standard logging; the trace() narration below is a separate channel
+logger = logging.getLogger(APP_NAME)
+
+
+@dataclass
+class RequestorState:
+    """What the main thread, the signal handler, and the API listener threads share: the
+    service, the requestor, and the shutdown event. The service and the requestor are
+    required, so every RequestorState holds both (built, though not necessarily connected
+    or started)."""
+    messaging_service: MessagingService
+    requestor: RequestReplyMessagePublisher
+    shutdown: threading.Event
+    connect_attempted: bool = False
+    exit_code: int = 0  # set to 1 on a failure exit (a missing reply, service interruption, or a failed terminate())
+
+
+def main(args: list[str]) -> int:
+    trace(f"{API} {APP_NAME} initializing...")
+    shutdown = threading.Event()
+
+    def on_shutdown_signal(signum: int, _frame) -> None:
+        trace(f"Shutdown signal received ({signal.Signals(signum).name}), stopping requestor...")
+        shutdown.set()
+
+    # graceful shutdown: SIGINT (Ctrl-C) or SIGTERM sets the shutdown event, no further
+    # request goes out, and teardown_solace() in the finally below stops the requestor and
+    # disconnects. Python runs signal handlers on the main thread, so the handler only
+    # flags; the cleanup runs on the main thread's normal exit path. Registered before
+    # connect, so a SIGTERM during the blocking connect() still reaches teardown.
+    signal.signal(signal.SIGINT, on_shutdown_signal)
+    signal.signal(signal.SIGTERM, on_shutdown_signal)
+    # basic username/password connection details, built by the shared connection-config
+    # helper: read from a config.json in the working directory if present, else from the
+    # command line (<host:port> <message-vpn> <client-username> [password]). Turning the
+    # application's input into Solace properties is the application's job, so it stays out
+    # of setup_solace(), which works from the properties alone.
+    properties = load_service_properties(args, APP_NAME)
+    # setup_solace() raises before any connection exists, so there is nothing to tear down
+    # if it fails
+    state = setup_solace(properties, shutdown)
+    try:
+        connect_solace(state)
+        send_requests(state)
+    finally:
+        teardown_solace(state)
+        trace("Main thread quitting.")
+    return state.exit_code  # non-zero after a failure, so scripts and supervisors see it
+
+
+def setup_solace(properties: dict, shutdown: threading.Event) -> RequestorState:
+    # build() creates the native session and resolves the host, so an unresolvable host
+    # fails here, before connect()
+    messaging_service = (
+        MessagingService.builder()
+        # the builder merges each call in order and a later call wins, so the reconnection
+        # strategy comes first: it is the default, and reconnection keys in the properties
+        # (for example from config.json) override it
+        .with_reconnection_retry_strategy(
+            RetryStrategy.parametrized_retry(RECONNECT_RETRIES, RECONNECT_RETRY_INTERVAL_MS))
+        .from_properties(properties)
+        .build()
+    )
+    # request-reply publisher: the API sets a unique reply-to topic on each request and
+    # correlates each reply to its request, so the requestor needs no reply subscription
+    # and no correlation id of its own. Its builder has no back-pressure setting (unlike
+    # the direct publisher), and its from_properties() applies no property, so there is
+    # nothing more to configure. Building the requestor needs only the built service;
+    # start() is what requires the connection.
+    requestor = (
+        messaging_service.request_reply()
+        .create_request_reply_message_publisher_builder()
+        .build()
+    )
+    state = RequestorState(messaging_service=messaging_service, requestor=requestor, shutdown=shutdown)
+    # best practice: register the service event listeners BEFORE connect(), so no
+    # reconnection or interruption event raised during or right after the connect is lost,
+    # and handle each event appropriately rather than only logging it
+    service_events = ServiceEventHandler(state)
+    messaging_service.add_reconnection_attempt_listener(service_events)
+    messaging_service.add_reconnection_listener(service_events)
+    messaging_service.add_service_interruption_listener(service_events)
+    return state
+
+
+def connect_solace(state: RequestorState) -> None:
+    state.connect_attempted = True
+    state.messaging_service.connect()  # blocking connect
+    state.requestor.start()  # raises IllegalStateError unless the service is connected
+
+
+def send_requests(state: RequestorState) -> None:
+    # the requests are OutboundMessages: publish() takes no bare payload, so the message
+    # builder is required here. One builder serves every request; the per-request
+    # difference goes in publish()'s additional_message_properties
+    message_builder = state.messaging_service.message_builder()
+    request_topic = Topic.of(REQUEST_TOPIC)
+    trace(f"{APP_NAME} sending {REQUEST_COUNT} direct requests to topic '{REQUEST_TOPIC}' "
+          f"(timeout {REQUEST_TIMEOUT_MS} ms each).")
+    # send phase: publish() returns a Future at once and the API waits for the reply on
+    # its own thread, so every request is in flight before the first reply is collected
+    pending: list[tuple[str, concurrent.futures.Future]] = []
+    for count in range(1, REQUEST_COUNT + 1):
+        if state.shutdown.is_set():
+            break  # a shutdown signal or a service interruption: send no further request
+        # an application message id lets the replier echo it onto the reply for
+        # traceability; the API does the request-reply correlation itself and does not
+        # need it
+        msg_id = str(uuid.uuid4())
+        # a bytearray payload travels as the binary attachment; the replier reads it back
+        # with get_payload_as_bytes()
+        request = message_builder.build(bytearray(f"Request {count} from {APP_NAME}".encode("utf-8")))
+        try:
+            future = state.requestor.publish(
+                request_message=request,
+                request_destination=request_topic,
+                reply_timeout=REQUEST_TIMEOUT_MS,
+                additional_message_properties={message_properties.APPLICATION_MESSAGE_ID: msg_id},
+            )
+        except PubSubPlusClientError as error:
+            # the requestor cannot take the request (for example it is terminating);
+            # retrying would not help
+            logger.error("Request %s could not be sent, quitting: %s", msg_id, error)
+            state.exit_code = 1
+            break
+        pending.append((msg_id, future))
+
+    # collect phase: each Future completes within REQUEST_TIMEOUT_MS of its send, with the
+    # reply or with the exception that ended the request. A Future can also take a
+    # callback, future.add_done_callback(fn), which runs on the API thread that completes
+    # it; the same do-not-block rule as the listeners applies there.
+    replies = 0
+    for msg_id, future in pending:
+        try:
+            reply = future.result()
+        except PubSubTimeoutError:
+            logger.warning("Request %s got no reply within %d ms.", msg_id, REQUEST_TIMEOUT_MS)
+            continue
+        except PubSubPlusClientError as error:
+            # the request could not be sent (for example the service is down)
+            logger.error("Request %s failed: %s", msg_id, error)
+            continue
+        replies += 1
+        # the replier sample sends a bytearray payload (the binary attachment), read back
+        # with get_payload_as_bytes(); a str payload arrives via get_payload_as_string()
+        # instead. errors="replace" keeps a non-UTF-8 payload from raising
+        payload = reply.get_payload_as_bytes()
+        text = payload.decode("utf-8", errors="replace") if payload is not None else reply.get_payload_as_string()
+        trace(f"{API} {APP_NAME} received a correlated reply (message id {reply.get_application_message_id()}): {text}")
+
+    trace(f"{API} {APP_NAME} received {replies} of {len(pending)} replies.")
+    # a request without its reply is a failure; a shutdown signal that stopped the send
+    # phase early is a clean stop, so only the requests actually sent count
+    if replies < len(pending):
+        state.exit_code = 1
+
+
+def teardown_solace(state: RequestorState) -> None:
+    # Application cleanup belongs here: once setup_solace() returns, teardown_solace() runs
+    # in main's finally on EVERY exit path (all replies, a missing reply, SIGINT/SIGTERM,
+    # service interruption, or an exception), so Solace teardown and application-side
+    # cleanup are never skipped.
+    state.shutdown.set()
+    if state.requestor.is_running():
+        # terminate() gives requests still waiting on their replies up to the grace period.
+        # It raises IncompleteMessageDeliveryError when requests are still outstanding after
+        # that; catch it so the disconnect below still runs
+        try:
+            state.requestor.terminate(TERMINATE_GRACE_PERIOD_MS)
+        except IncompleteMessageDeliveryError as error:
+            logger.error("Requestor stopped with requests still outstanding after %d ms: %s",
+                         TERMINATE_GRACE_PERIOD_MS, error)
+            state.exit_code = 1
+        except PubSubPlusClientError as error:
+            logger.error("Requestor terminate() failed: %s", error)
+            state.exit_code = 1
+    if state.connect_attempted:
+        # disconnect() raises IllegalStateError on a service that never attempted to connect,
+        # hence the flag; on a service that is already down it returns quietly
+        state.messaging_service.disconnect()
+
+
+def trace(message: str) -> None:
+    """Demo narration sink: every status line in this sample funnels through this one
+    function. An application replaces this single body to route narration to its
+    logger or reporting system. The logger calls for API events (service, request)
+    are a separate channel and stay as they are."""
+    print(message, flush=True)
+
+
+############################################################################
+
+
+class ServiceEventHandler(ReconnectionAttemptListener, ReconnectionListener, ServiceInterruptionListener):
+    """The three service event listeners in one class. Callbacks run on an API thread:
+    return promptly and do not block in them, since events and delivery stall while a
+    callback runs (C API Best Practices)."""
+
+    def __init__(self, state: RequestorState) -> None:
+        self.state = state
+
+    def on_reconnecting(self, event: ServiceEvent) -> None:
+        # connection lost, automatic reconnect attempt in progress. Request-reply is direct
+        # (at-most-once): a request or reply in flight during the outage can be lost, and
+        # that request then times out
+        logger.warning("Service reconnecting; requests in flight can time out: %s (%s)",
+                       event.get_message(), event.get_cause())
+
+    def on_reconnected(self, event: ServiceEvent) -> None:
+        # automatic reconnect succeeded: requests can go out again
+        logger.info("Service reconnected to %s", event.get_broker_uri())
+
+    def on_service_interrupted(self, event: ServiceEvent) -> None:
+        # the connection went down and cannot be restored (reconnect attempts exhausted)
+        logger.error("Service interrupted and cannot be restored, quitting: %s (%s)",
+                     event.get_message(), event.get_cause())
+        # Application cleanup signal: the service will not recover. Trigger application-side
+        # cleanup from here. This sample sets shutdown, so no further request goes out, the
+        # requests in flight fail or time out, and teardown_solace() runs in main's finally.
+        self.state.exit_code = 1
+        self.state.shutdown.set()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+    sys.exit(main(sys.argv[1:]))
diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-sample.py
similarity index 99%
rename from plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-async-sample.py
rename to plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-sample.py
index 4f1319c..3da08ee 100644
--- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-async-sample.py
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-direct-subscriber-sample.py
@@ -28,7 +28,7 @@
 
 Only practices documented in canonical Solace sources are encoded here.
 
-Copied into a generated project as direct_subscriber_async.py, next to the shared helper
+Copied into a generated project as direct_subscriber.py, next to the shared helper
 python-solace-connection-config.py copied as solace_connection_config.py.
 
 Any generated adaptation of this sample MUST begin with the exact line:
diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-sample.py
similarity index 97%
rename from plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-async-sample.py
rename to plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-sample.py
index a94e4b5..429babf 100644
--- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-async-sample.py
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-publisher-sample.py
@@ -28,7 +28,7 @@
 
 Only practices documented in canonical Solace sources are encoded here.
 
-Copied into a generated project as guaranteed_publisher_async.py, next to the shared
+Copied into a generated project as guaranteed_publisher.py, next to the shared
 helper python-solace-connection-config.py copied as solace_connection_config.py.
 
 Any generated adaptation of this sample MUST begin with the exact line:
@@ -236,10 +236,10 @@ def run_publish_loop(state: PublisherState) -> None:
             # publish() blocks while the buffer is full (back pressure). During an automatic
             # reconnect the messages wait in the buffer for the transport and go out after
             # it, and publish() blocks only once the buffer fills; the loop needs no pause of
-            # its own (the direct publisher, whose messages fail instead of waiting, does need
-            # one). A shutdown signal that
-            # arrives during that wait takes effect once publish() returns: when the buffer
-            # drains, or when the service goes down and the API releases the wait.
+            # its own (in the direct publisher, those messages fail instead of waiting). A
+            # shutdown signal that arrives during that wait takes effect once publish()
+            # returns: when the buffer drains, or when the service goes down and the API
+            # releases the wait.
             state.publisher.publish(message, topic, user_context=msg_id)
             state.msg_sent_counter += 1
         except PubSubPlusClientError as error:
diff --git a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-async-sample.py b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-sample.py
similarity index 99%
rename from plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-async-sample.py
rename to plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-sample.py
index dd0936e..0c67bde 100644
--- a/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-async-sample.py
+++ b/plugins/solace-messaging-skills/skills/solace-application-development/references/python/python-guaranteed-subscriber-sample.py
@@ -29,7 +29,7 @@
 
 Only practices documented in canonical Solace sources are encoded here.
 
-Copied into a generated project as guaranteed_subscriber_async.py, next to the shared
+Copied into a generated project as guaranteed_subscriber.py, next to the shared
 helper python-solace-connection-config.py copied as solace_connection_config.py.
 
 Any generated adaptation of this sample MUST begin with the exact line: