.

listener.py

@@ -2,12 +2,8 @@ import json
 import aio_pika
 from typing import Callable, Awaitable, Dict, Any, List, Optional
 
-from models import CloudEvent
-
-Handler = Callable[[Any], Awaitable[None]]  # data can be dict / list / str
-
 class RabbitListenerBase:
-    def __init__(self, base, instance_name: str, handlers: Dict[str, Handler]):
+    def __init__(self, base, instance_name: str, handlers: Dict[str, Callable[[Any], Awaitable[None]]]):
         self._base = base
         self._instance_name = instance_name
         self._handlers = handlers
@@ -35,13 +31,11 @@ class RabbitListenerBase:
             async with msg.process():
                 try:
                     envelope = json.loads(msg.body.decode("utf-8"))
-                    # Validate basic CloudEvent shape without being strict
-                    # (C# side doesn’t require strict validation either)
-                    data = envelope.get("data", None)
+                    data = envelope.get("data", None)  # dict / list / str
                     if handler:
                         await handler(data)
                 except Exception:
-                    # swallow to avoid redelivery storms; log if you wire a logger
+                    # Avoid requeue storms; log if you add a logger
                     pass
 
         return _on_msg

models.py

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 from typing import Any, Optional, List
 from datetime import datetime, timezone
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ConfigDict
 
 
-# ---------- CloudEvent ----------
+# ---------- CloudEvent (Pydantic v2 model) ----------
 class CloudEvent(BaseModel):
     specversion: str = "1.0"
     id: str
@@ -14,7 +16,7 @@ class CloudEvent(BaseModel):
     data: Optional[Any] = None
 
     @staticmethod
-    def now_iso() -> datetime:
+    def now_utc() -> datetime:
         return datetime.now(timezone.utc)
 
     @classmethod
@@ -23,17 +25,11 @@ class CloudEvent(BaseModel):
             id=event_id,
             type=event_type or ("NullOrEmpty" if data is None else type(data).__name__),
             source=source,
-            time=cls.now_iso(),
+            time=cls.now_utc(),
             data=data,
         )
 
 
-from typing import Any, Dict, Optional, List
-from datetime import datetime, timezone
-from pydantic import BaseModel, Field, ConfigDict  # <-- add ConfigDict
-
-# ... keep CloudEvent as you had it ...
-
 # ---------- Permissive ancillary types ----------
 class FunctionState(BaseModel):
     IsFunctionCall: bool = False
@@ -42,19 +38,13 @@ class FunctionState(BaseModel):
     IsFunctionCallStatus: bool = False
     IsFunctionStillRunning: bool = False
 
-    def set_tuple(self, call: bool, resp: bool, err: bool, status: bool, running: bool):
-        self.IsFunctionCall = call
-        self.IsFunctionCallResponse = resp
-        self.IsFunctionCallError = err
-        self.IsFunctionCallStatus = status
-        self.IsFunctionStillRunning = running
 
 class FunctionCallData(BaseModel):
-    # Accept any extra keys (v2 style)
+    # Accept arbitrary keys so we mirror your .NET payloads without schema fights
     model_config = ConfigDict(extra="allow")
 
+
 class UserInfo(BaseModel):
-    # Accept any extra keys (v2 style)
     model_config = ConfigDict(extra="allow")
 
 
@@ -93,57 +83,20 @@ class LLMServiceObj(BaseModel):
     UserInfo: UserInfo = Field(default_factory=UserInfo)
     StartTimeUTC: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
 
-    # stacks (serialize as arrays; C# Stack<T> will read fine)
+    # stacks (arrays; C# Stack<T> will deserialize fine)
     LlmStack: List[str] = Field(default_factory=list)
     FunctionCallIdStack: List[str] = Field(default_factory=list)
     FunctionNameStack: List[str] = Field(default_factory=list)
     IsProcessedStack: List[bool] = Field(default_factory=list)
     MessageIDStack: List[str] = Field(default_factory=list)
 
-    # function state (maps to your FunctionState object)
-    # Your C# stores this in a private field + setters; we expose the same booleans as a nested object.
+    # function state flags (since your C# exposes these via methods/flags)
     IsFunctionCall: bool = False
     IsFunctionCallResponse: bool = False
     IsFunctionCallError: bool = False
     IsFunctionCallStatus: bool = False
     IsFunctionStillRunning: bool = False
 
-    # Convenience helpers (optional)
-    def push_llm(self, llm_name: str, new_call_id: str, new_func_name: str, new_message_id: str, new_is_processed: bool):
-        if self.SourceLlm:
-            self.LlmStack.append(self.SourceLlm)
-        self.SourceLlm = self.DestinationLlm
-        self.DestinationLlm = llm_name
-
-        if self.MessageID:
-            self.MessageIDStack.append(self.MessageID)
-        self.MessageID = new_message_id
-
-        if self.FunctionCallId:
-            self.FunctionCallIdStack.append(self.FunctionCallId)
-        self.FunctionCallId = new_call_id
-
-        if self.FunctionName:
-            self.FunctionNameStack.append(self.FunctionName)
-        self.FunctionName = new_func_name
-
-        self.IsProcessedStack.append(self.IsProcessed)
-        self.IsProcessed = new_is_processed
-
-    def pop_llm(self):
-        if self.LlmStack:
-            self.SourceLlm = self.LlmStack.pop()
-        self.DestinationLlm = self.SourceLlm
-
-        if self.MessageIDStack:
-            self.MessageID = self.MessageIDStack.pop()
-        if self.FunctionCallIdStack:
-            self.FunctionCallId = self.FunctionCallIdStack.pop()
-        if self.FunctionNameStack:
-            self.FunctionName = self.FunctionNameStack.pop()
-        if self.IsProcessedStack:
-            self.IsProcessed = self.IsProcessedStack.pop()
-
 
 # ---------- ResultObj ----------
 class ResultObj(BaseModel):

rabbit_base.py

@@ -1,10 +1,12 @@
-import asyncio, json, uuid
-import aio_pika
+import asyncio
 from typing import Callable, Dict, List, Optional
+import aio_pika
+
 from config import settings
 
 ExchangeResolver = Callable[[str], str]  # exchangeName -> exchangeType
 
+
 class RabbitBase:
     def __init__(self, exchange_type_resolver: Optional[ExchangeResolver] = None):
         self._conn: Optional[aio_pika.RobustConnection] = None

rabbit_repo.py

@@ -2,14 +2,16 @@ import uuid
 from typing import Any, Optional
 import aio_pika
 
+from config import settings
 from models import CloudEvent
+from rabbit_base import RabbitBase
 from utils import to_json, json_compress_str
 
 
-class RabbitRepo:
-     def __init__(self, external_source: str):
+class RabbitRepo(RabbitBase):
+    def __init__(self, external_source: str):
         super().__init__(exchange_type_resolver=self._resolve_type)
-        self._external_source = external_source  # like SystemUrl.ExternalUrl
+        self._source = external_source  # like SystemUrl.ExternalUrl
 
     def _resolve_type(self, exch: str) -> str:
         # longest prefix wins (like your .NET mapping)
@@ -19,19 +21,19 @@ class RabbitRepo:
         return settings.RABBIT_EXCHANGE_TYPE
 
     async def publish(self, exchange: str, obj: Any, routing_key: str = ""):
-        ex = await self._base.ensure_exchange(exchange)
+        ex = await self.ensure_exchange(exchange)
         evt = CloudEvent.wrap(
             event_id=str(uuid.uuid4()),
             event_type=(obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
             source=self._source,
-            data=obj,
+            data=obj if not hasattr(obj, "model_dump") else obj.model_dump(),
         )
-        body = evt.model_dump_json(by_alias=False, exclude_none=True).encode("utf-8")
+        body = evt.model_dump_json(exclude_none=True).encode("utf-8")
         await ex.publish(aio_pika.Message(body=body), routing_key=routing_key)
 
     async def publish_jsonz(self, exchange: str, obj: Any, routing_key: str = "", with_id: Optional[str] = None) -> str:
-        ex = await self._base.ensure_exchange(exchange)
-        datajson = to_json(obj)
+        ex = await self.ensure_exchange(exchange)
+        datajson = to_json(obj if not hasattr(obj, "model_dump") else obj.model_dump())
         datajsonZ = json_compress_str(datajson)
         payload: Any = (datajsonZ, with_id) if with_id else datajsonZ
 
@@ -41,6 +43,6 @@ class RabbitRepo:
             source=self._source,
             data=payload,
         )
-        body = evt.model_dump_json(by_alias=False, exclude_none=True).encode("utf-8")
+        body = evt.model_dump_json(exclude_none=True).encode("utf-8")
         await ex.publish(aio_pika.Message(body=body), routing_key=routing_key)
         return datajsonZ

requirements.txt

@@ -4,3 +4,4 @@ uvicorn==0.35.0
 aio-pika==9.5.7
 pydantic==2.11.1
 pydantic-settings==2.10.1
+