.

app.py

@@ -2,7 +2,9 @@
 import asyncio, logging, gradio as gr
 from config import settings
 from rabbit_base import RabbitBase
-from openai_server import ChatCompletionsServer, ImagesServer
+from listener import RabbitListenerBase
+from rabbit_repo import RabbitRepo
+from oa_server import OpenAIServers
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
 log = logging.getLogger("app")
@@ -14,27 +16,45 @@ try:
 except Exception:
     def gpu_entrypoint() -> str: return "gpu: not available (CPU only)"
 
-# Ensure oa.* exchanges are DIRECT; fall back to your global default otherwise.
-resolver = lambda name: ("direct" if name.startswith("oa.") else settings.RABBIT_EXCHANGE_TYPE)
+# publisher + servers
+publisher = RabbitRepo(external_source="openai.mq.server")
+# Force oa.* exchanges to DIRECT using the built-in resolver hook (no raw API added)
+resolver = (lambda name: "direct" if name.startswith("oa.") else settings.RABBIT_EXCHANGE_TYPE)
 base = RabbitBase(exchange_type_resolver=resolver)
 
-chat_srv = ChatCompletionsServer(base, exchange_name="oa.chat.create", routing_key="default")
-img_srv  = ImagesServer(base,         exchange_name="oa.images.generate", routing_key="default")
+servers = OpenAIServers(publisher)
+
+# Existing handlers can stay; add our two:
+handlers = {
+    "oaChatCreate": servers.handle_chat_create,
+    "oaImagesGenerate": servers.handle_images_generate,
+}
+
+# Declare listener queues using your proven pattern
+DECLS = [
+    # Chat Completions
+    {"ExchangeName": "oa.chat.create", "FuncName": "oaChatCreate",
+     "MessageTimeout": 600_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+    # Images (generations)
+    {"ExchangeName": "oa.images.generate", "FuncName": "oaImagesGenerate",
+     "MessageTimeout": 600_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+]
+
+listener = RabbitListenerBase(base, instance_name=settings.RABBIT_INSTANCE_NAME, handlers=handlers)
 
 async def _startup_init():
     try:
-        # This connect() uses your TLS settings (verify disabled if amqps)
-        await base.connect()
-        await asyncio.gather(chat_srv.start(), img_srv.start())
+        await base.connect()          # your TLS settings apply here
+        await listener.start(DECLS)   # same listener pattern as before
         return "OpenAI MQ servers: ready"
     except Exception as e:
         log.exception("Startup init failed")
         return f"ERROR: {e}"
 
-async def ping() -> str: return "ok"
+async def ping(): return "ok"
 
-with gr.Blocks(title="OpenAI over RabbitMQ", theme=gr.themes.Soft()) as demo:
-    gr.Markdown("## OpenAI-compatible server over RabbitMQ (using RabbitBase TLS settings)")
+with gr.Blocks(title="OpenAI over RabbitMQ (proven API)", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("## OpenAI-compatible over RabbitMQ — using existing Rabbit API (CloudEvent envelopes)")
     with gr.Tabs():
         with gr.Tab("Service"):
             btn = gr.Button("Ping"); out = gr.Textbox(label="Ping result")

oa_server.py

@@ -0,0 +1,111 @@
+# oa_server.py
+from __future__ import annotations
+import json, time, uuid, logging
+from typing import Any, Dict, List, AsyncIterable, Optional
+
+from rabbit_repo import RabbitRepo
+
+logger = logging.getLogger(__name__)
+
+# ------------------ helpers ------------------
+def _now() -> int: return int(time.time())
+def _chunk_text(s: str, sz: int = 140) -> List[str]:
+    return [s[i:i+sz] for i in range(0, len(s or ""), sz)] if s else []
+def _last_user_text(messages: List[Dict[str, Any]]) -> str:
+    for m in reversed(messages or []):
+        if (m or {}).get("role") == "user":
+            c = m.get("content", "")
+            if isinstance(c, str):
+                return c
+            if isinstance(c, list):
+                texts = [p.get("text","") for p in c if p.get("type") == "text"]
+                return " ".join([t for t in texts if t])
+    return ""
+
+# ------------------ backends (replace later) ------------------
+class ChatBackend:
+    async def stream(self, request: Dict[str, Any]) -> AsyncIterable[Dict[str, Any]]:
+        raise NotImplementedError
+
+class DummyChatBackend(ChatBackend):
+    async def stream(self, request: Dict[str, Any]) -> AsyncIterable[Dict[str, Any]]:
+        rid   = f"chatcmpl-{uuid.uuid4().hex[:12]}"
+        model = request.get("model", "gpt-4o-mini")
+        text  = _last_user_text(request.get("messages", [])) or "(empty)"
+        out   = f"Echo (Rabbit): {text}"
+        now   = _now()
+
+        # role delta
+        yield {"id": rid, "object":"chat.completion.chunk", "created": now, "model": model,
+               "choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":None}]}
+        # content deltas
+        for piece in _chunk_text(out, 140):
+            yield {"id": rid, "object":"chat.completion.chunk", "created": now, "model": model,
+                   "choices":[{"index":0,"delta":{"content":piece},"finish_reason":None}]}
+        # final delta
+        yield {"id": rid, "object":"chat.completion.chunk", "created": now, "model": model,
+               "choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
+
+class ImagesBackend:
+    async def generate_b64(self, request: Dict[str, Any]) -> str:
+        # 1x1 transparent PNG (stub)
+        return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4nGP4BwQACfsD/etCJH0AAAAASUVORK5CYII="
+
+# ------------------ handler class ------------------
+class OpenAIServers:
+    """
+    Handlers you can register in RabbitListenerBase:
+      - 'oaChatCreate'       -> handle_chat_create
+      - 'oaImagesGenerate'   -> handle_images_generate
+    Uses RabbitRepo.publish(...) to emit CloudEvent-wrapped OpenAI JSON.
+    """
+    def __init__(self, publisher: RabbitRepo,
+                 *, chat_backend: Optional[ChatBackend] = None,
+                    images_backend: Optional[ImagesBackend] = None):
+        self._pub = publisher
+        self._chat = chat_backend or DummyChatBackend()
+        self._img  = images_backend or ImagesBackend()
+
+    # -------- Chat Completions --------
+    async def handle_chat_create(self, data: Dict[str, Any]) -> None:
+        """
+        data: OpenAI chat request + 'reply_key' (string)
+        Server publishes to exchange 'oa.chat.reply' with routing_key = reply_key.
+        """
+        if not isinstance(data, dict):
+            logger.warning("oaChatCreate: data is not a dict")
+            return
+
+        reply_key = data.get("reply_key")
+        if not reply_key:
+            logger.error("oaChatCreate: missing reply_key")
+            return
+
+        try:
+            async for chunk in self._chat.stream(data):
+                # CloudEvent-wrapped OpenAI chunk to oa.chat.reply
+                await self._pub.publish("oa.chat.reply", chunk, routing_key=reply_key)
+            # Optional sentinel
+            await self._pub.publish("oa.chat.reply", {"object": "stream.end"}, routing_key=reply_key)
+        except Exception:
+            logger.exception("oaChatCreate: streaming failed")
+
+    # -------- Images (generations) --------
+    async def handle_images_generate(self, data: Dict[str, Any]) -> None:
+        """
+        data: OpenAI images.generate request + 'reply_key' (string)
+        """
+        if not isinstance(data, dict):
+            logger.warning("oaImagesGenerate: data is not a dict")
+            return
+        reply_key = data.get("reply_key")
+        if not reply_key:
+            logger.error("oaImagesGenerate: missing reply_key")
+            return
+
+        try:
+            b64 = await self._img.generate_b64(data)
+            resp = {"created": _now(), "data":[{"b64_json": b64}]}
+            await self._pub.publish("oa.images.reply", resp, routing_key=reply_key)
+        except Exception:
+            logger.exception("oaImagesGenerate: generation failed")

openai_server.py

@@ -1,137 +0,0 @@
-# openai_server.py
-from __future__ import annotations
-import json, time, uuid, logging
-from typing import Any, Dict, List, Optional, AsyncIterable
-import aio_pika
-from rabbit_base import RabbitBase
-
-logger = logging.getLogger(__name__)
-
-def _now() -> int: return int(time.time())
-def _chunk_text(s: str, sz: int = 140) -> List[str]:
-    return [s[i:i+sz] for i in range(0, len(s or ""), sz)] if s else []
-def _last_user_text(messages: List[Dict[str, Any]]) -> str:
-    for m in reversed(messages or []):
-        if (m or {}).get("role") == "user":
-            c = m.get("content", "")
-            if isinstance(c, str):
-                return c
-            if isinstance(c, list):
-                texts = [p.get("text","") for p in c if p.get("type") == "text"]
-                return " ".join([t for t in texts if t])
-    return ""
-
-# ------------ Backends ------------
-class ChatBackend:
-    async def stream(self, request: Dict[str, Any]) -> AsyncIterable[Dict[str, Any]]:
-        raise NotImplementedError
-
-class DummyChatBackend(ChatBackend):
-    async def stream(self, request: Dict[str, Any]) -> AsyncIterable[Dict[str, Any]]:
-        rid = f"chatcmpl-{uuid.uuid4().hex[:12]}"
-        model = request.get("model", "gpt-4o-mini")
-        text = _last_user_text(request.get("messages", [])) or "(empty)"
-        answer = f"Echo (RabbitMQ): {text}"
-        now = _now()
-        # role delta first
-        yield {"id": rid,"object":"chat.completion.chunk","created":now,"model":model,
-               "choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":None}]}
-        # content deltas
-        for piece in _chunk_text(answer, 140):
-            yield {"id": rid,"object":"chat.completion.chunk","created":now,"model":model,
-                   "choices":[{"index":0,"delta":{"content":piece},"finish_reason":None}]}
-        # final
-        yield {"id": rid,"object":"chat.completion.chunk","created":now,"model":model,
-               "choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}
-
-class ImagesBackend:
-    async def generate_b64(self, request: Dict[str, Any]) -> str:
-        # 1x1 transparent PNG
-        return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4nGP4BwQACfsD/etCJH0AAAAASUVORK5CYII="
-
-# ------------ Servers using RabbitBase ------------
-class ChatCompletionsServer:
-    def __init__(self, base: RabbitBase, *, exchange_name="oa.chat.create", routing_key="default",
-                 backend: Optional[ChatBackend] = None):
-        self._base = base
-        self._exchange = exchange_name
-        self._routing_key = routing_key
-        self._backend = backend or DummyChatBackend()
-        self._queue_name = f"{exchange_name}.{routing_key}"
-
-    async def start(self):
-        # declare exchange + queue with *your* TLS/connection behavior
-        q = await self._base.declare_queue_bind(
-            exchange=self._exchange,
-            queue_name=self._queue_name,
-            routing_keys=[self._routing_key],
-            ttl_ms=None,
-        )
-        await q.consume(self._on_message)
-        logger.info("ChatCompletionsServer listening on %s/%s → %s",
-                    self._exchange, self._routing_key, self._queue_name)
-
-    async def _on_message(self, msg: aio_pika.IncomingMessage):
-        async with msg.process(ignore_processed=True):
-            try:
-                req = json.loads(msg.body.decode("utf-8", errors="replace"))
-                reply_to = msg.reply_to
-                corr_id  = msg.correlation_id
-                if not reply_to or not corr_id:
-                    logger.warning("Missing reply_to/correlation_id; dropping.")
-                    return
-
-                async for chunk in self._backend.stream(req):
-                    await self._base.publish_basic(
-                        routing_key=reply_to,
-                        body=json.dumps(chunk).encode("utf-8"),
-                        correlation_id=corr_id,
-                    )
-                # optional end sentinel
-                await self._base.publish_basic(
-                    routing_key=reply_to,
-                    body=b'{"object":"stream.end"}',
-                    correlation_id=corr_id,
-                )
-            except Exception:
-                logger.exception("ChatCompletionsServer: failed to process message")
-
-class ImagesServer:
-    def __init__(self, base: RabbitBase, *, exchange_name="oa.images.generate", routing_key="default",
-                 backend: Optional[ImagesBackend] = None):
-        self._base = base
-        self._exchange = exchange_name
-        self._routing_key = routing_key
-        self._backend = backend or ImagesBackend()
-        self._queue_name = f"{exchange_name}.{routing_key}"
-
-    async def start(self):
-        q = await self._base.declare_queue_bind(
-            exchange=self._exchange,
-            queue_name=self._queue_name,
-            routing_keys=[self._routing_key],
-            ttl_ms=None,
-        )
-        await q.consume(self._on_message)
-        logger.info("ImagesServer listening on %s/%s → %s",
-                    self._exchange, self._routing_key, self._queue_name)
-
-    async def _on_message(self, msg: aio_pika.IncomingMessage):
-        async with msg.process(ignore_processed=True):
-            try:
-                req = json.loads(msg.body.decode("utf-8", errors="replace"))
-                reply_to = msg.reply_to
-                corr_id  = msg.correlation_id
-                if not reply_to or not corr_id:
-                    logger.warning("Missing reply_to/correlation_id; dropping.")
-                    return
-
-                b64 = await self._backend.generate_b64(req)
-                resp = {"created": _now(), "data":[{"b64_json": b64}]}
-                await self._base.publish_basic(
-                    routing_key=reply_to,
-                    body=json.dumps(resp).encode("utf-8"),
-                    correlation_id=corr_id,
-                )
-            except Exception:
-                logger.exception("ImagesServer: failed to process message")