.

listener.py

@@ -1,47 +1,47 @@
 import json
-from typing import Callable, Dict, List, Optional
 import aio_pika
-from rabbit_base import RabbitBase
-from config import settings
+from typing import Callable, Awaitable, Dict, Any, List, Optional
 
-# Maps FuncName -> handler coroutine
-Handler = Callable[[dict], "awaitable[None]"]
+from models import CloudEvent
 
-class RabbitListenerBase(RabbitBase):
-    def __init__(self, service_id: str, handlers: Dict[str, Handler]):
-        super().__init__()
-        self._service_id = service_id
+Handler = Callable[[Any], Awaitable[None]]  # data can be dict / list / str
+
+class RabbitListenerBase:
+    def __init__(self, base, instance_name: str, handlers: Dict[str, Handler]):
+        self._base = base
+        self._instance_name = instance_name
         self._handlers = handlers
         self._consumers: List[aio_pika.abc.AbstractRobustQueue] = []
 
     def _qname(self, exchange: str, routing_keys: List[str]) -> str:
-        rk_part = "-".join(sorted([rk for rk in (routing_keys or [""]) if rk != ""])) or ""
+        rk_part = "-".join(sorted([rk for rk in (routing_keys or [""]) if rk])) or ""
         suffix = f"-{rk_part}" if rk_part else ""
-        return f"{settings.RABBIT_INSTANCE_NAME}-{exchange}{suffix}"
+        return f"{self._instance_name}-{exchange}{suffix}"
 
     async def start(self, declarations: List[dict]):
-        """
-        declarations: list of {ExchangeName, FuncName, MessageTimeout, Type?, RoutingKeys?}
-        """
         for d in declarations:
             exch = d["ExchangeName"]
-            rks = d.get("RoutingKeys") or [settings.RABBIT_ROUTING_KEY]
             ttl = d.get("MessageTimeout") or None
-            q = await self.declare_queue_bind(exchange=exch, queue_name=self._qname(exch, rks), routing_keys=rks, ttl_ms=ttl)
+            rks = d.get("RoutingKeys") or [""]
+            qname = self._qname(exch, rks)
+            q = await self._base.declare_queue_bind(exchange=exch, queue_name=qname, routing_keys=rks, ttl_ms=ttl)
             await q.consume(self._make_consumer(d["FuncName"]))
             self._consumers.append(q)
 
     def _make_consumer(self, func_name: str):
         handler = self._handlers.get(func_name)
+
         async def _on_msg(msg: aio_pika.IncomingMessage):
             async with msg.process():
                 try:
-                    # Expect CloudEvent JSON
                     envelope = json.loads(msg.body.decode("utf-8"))
-                    data = envelope.get("data")
+                    # Validate basic CloudEvent shape without being strict
+                    # (C# side doesn’t require strict validation either)
+                    data = envelope.get("data", None)
                     if handler:
                         await handler(data)
-                except Exception as e:
-                    # swallow to avoid nack loops; your logger can capture details
+                except Exception:
+                    # swallow to avoid redelivery storms; log if you wire a logger
                     pass
+
         return _on_msg

models.py

@@ -0,0 +1,149 @@
+from typing import Any, Optional, List
+from datetime import datetime, timezone
+from pydantic import BaseModel, Field
+
+
+# ---------- CloudEvent ----------
+class CloudEvent(BaseModel):
+    specversion: str = "1.0"
+    id: str
+    type: str
+    source: str
+    time: datetime
+    datacontenttype: str = "application/json"
+    data: Optional[Any] = None
+
+    @staticmethod
+    def now_iso() -> datetime:
+        return datetime.now(timezone.utc)
+
+    @classmethod
+    def wrap(cls, *, event_id: str, event_type: str, source: str, data: Any) -> "CloudEvent":
+        return cls(
+            id=event_id,
+            type=event_type or ("NullOrEmpty" if data is None else type(data).__name__),
+            source=source,
+            time=cls.now_iso(),
+            data=data,
+        )
+
+
+# ---------- Permissive ancillary types you referenced ----------
+class FunctionState(BaseModel):
+    IsFunctionCall: bool = False
+    IsFunctionCallResponse: bool = False
+    IsFunctionCallError: bool = False
+    IsFunctionCallStatus: bool = False
+    IsFunctionStillRunning: bool = False
+
+    # Convenience to set the 5-tuple like your C# SetFunctionState
+    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):
+    # Add fields as you need; permissive placeholder
+    __root__: dict = Field(default_factory=dict)
+
+
+class UserInfo(BaseModel):
+    # Add fields as you need; permissive placeholder
+    __root__: dict = Field(default_factory=dict)
+
+
+# ---------- LLMServiceObj (field names match C# exactly) ----------
+class LLMServiceObj(BaseModel):
+    # strings
+    SessionId: str = ""
+    JsonFunction: str = ""
+    LlmMessage: str = ""
+    ResultMessage: str = ""
+    UserInput: str = ""
+    RequestSessionId: str = ""
+    FunctionName: str = ""
+    TimeZone: str = ""
+    LLMRunnerType: str = "TurboLLM"
+    SourceLlm: str = ""
+    DestinationLlm: str = ""
+    MessageID: str = ""
+    LlmSessionStartName: str = ""
+    SwapFunctionName: str = ""
+    ChatAgentLocation: str = ""
+    ToolsDefinitionId: Optional[str] = None
+    JsonToolsBuilderSpec: Optional[str] = None
+
+    # ints / bools
+    TokensUsed: int = 0
+    IsUserLoggedIn: bool = False
+    IsFuncAck: bool = False
+    IsProcessed: bool = False
+    IsSystemLlm: bool = False
+    Timeout: Optional[int] = None
+
+    # complex
+    FunctionCallId: str = ""
+    FunctionCallData: FunctionCallData = Field(default_factory=FunctionCallData)
+    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)
+    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.
+    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):
+    Message: str = ""
+    Success: bool = False
+    Data: Optional[Any] = None

rabbit_repo.py

@@ -1,12 +1,13 @@
 import uuid
-from typing import Any
-from rabbit_base import RabbitBase
-from cloud_event import CloudEvent
-from config import settings
+from typing import Any, Optional
+import aio_pika
+
+from models import CloudEvent
 from utils import to_json, json_compress_str
 
-class RabbitRepo(RabbitBase):
-    def __init__(self, external_source: str):
+
+class RabbitRepo:
+     def __init__(self, external_source: str):
         super().__init__(exchange_type_resolver=self._resolve_type)
         self._external_source = external_source  # like SystemUrl.ExternalUrl
 
@@ -18,17 +19,28 @@ class RabbitRepo(RabbitBase):
         return settings.RABBIT_EXCHANGE_TYPE
 
     async def publish(self, exchange: str, obj: Any, routing_key: str = ""):
-        ex = await self.ensure_exchange(exchange)
-        payload = CloudEvent.wrap(obj, event_type=(obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
-                                  source=self._external_source, id=str(uuid.uuid4()))
-        await ex.publish(aio_pika.Message(body=payload), routing_key=routing_key)
+        ex = await self._base.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,
+        )
+        body = evt.model_dump_json(by_alias=False, 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)
+        datajsonZ = json_compress_str(datajson)
+        payload: Any = (datajsonZ, with_id) if with_id else datajsonZ
 
-    async def publish_jsonz(self, exchange: str, obj: Any, routing_key: str = "", with_id: str | None = None):
-        ex = await self.ensure_exchange(exchange)
-        json_str = to_json(obj)
-        datajsonZ = json_compress_str(json_str)
-        to_send = (datajsonZ, with_id) if with_id else datajsonZ
-        payload = CloudEvent.wrap(to_send, event_type=(obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
-                                  source=self._external_source, id=str(uuid.uuid4()))
-        await ex.publish(aio_pika.Message(body=payload), routing_key=routing_key)
+        evt = CloudEvent.wrap(
+            event_id=str(uuid.uuid4()),
+            event_type=(obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
+            source=self._source,
+            data=payload,
+        )
+        body = evt.model_dump_json(by_alias=False, exclude_none=True).encode("utf-8")
+        await ex.publish(aio_pika.Message(body=body), routing_key=routing_key)
         return datajsonZ

runners/service.py

@@ -1,127 +1,247 @@
+# service.py
 import asyncio
-from typing import Dict, Optional
-from collections import defaultdict
+from dataclasses import dataclass
+from typing import Any, Dict, Optional
 
-from rabbit_repo import RabbitRepo
 from config import settings
+from models import LLMServiceObj, ResultObj
+from rabbit_repo import RabbitRepo
 from runners.base import ILLMRunner
 
+
+@dataclass
+class _Session:
+    Runner: Optional[ILLMRunner]
+    FullSessionId: str
+
+
 class LLMService:
+    """
+    Python/Gradio equivalent of your .NET LLMService.
+    Keeps identical field names and queue semantics when talking to RabbitMQ.
+    """
     def __init__(self, publisher: RabbitRepo, runner_factory):
         self._pub = publisher
-        self._runner_factory = runner_factory
-        self._sessions: Dict[str, dict] = {}  # sessionId -> {"Runner": ILLMRunner, "FullSessionId": str}
+        self._runner_factory = runner_factory  # async factory: dict|LLMServiceObj -> ILLMRunner
+        self._sessions: Dict[str, _Session] = {}
         self._ready = asyncio.Event()
-        self._ready.set()  # if you have async load, clear and set after
+        # If you need async load (history, etc.), call self._ready.clear() and later set
+        self._ready.set()
 
     async def init(self):
-        # If you have history to load, do here then self._ready.set()
+        """
+        Hook to preload history/sessions if needed; call self._ready.set() when finished.
+        """
+        # Example:
+        # self._ready.clear()
+        # await load_history()
+        # self._ready.set()
         pass
 
-    async def _set_result(self, obj: dict, message: str, success: bool, queue: str, check_system: bool=False):
-        obj["ResultMessage"] = message
-        obj["ResultSuccess"] = success
-        obj["LlmMessage"] = (f"<Success>{message}</Success>" if success else f"<Error>{message}</Error>")
-        # mirror your .NET rule (don’t publish for system llm if check_system is True)
-        if not (check_system and obj.get("IsSystemLlm")):
-            await self._pub.publish(queue, obj)
+    # ---------------------------- helpers ----------------------------
+
+    def _to_model(self, data: Any) -> LLMServiceObj:
+        """
+        Accepts LLMServiceObj or dict and returns a validated LLMServiceObj.
+        """
+        if isinstance(data, LLMServiceObj):
+            return data
+        if isinstance(data, dict):
+            return LLMServiceObj(**data)
+        # If your pipeline ever sends compressed strings here, decompress+parse first.
+        raise TypeError("LLMService expects an object payload (dict/LLMServiceObj).")
+
+    async def _emit_result(
+        self,
+        obj: LLMServiceObj,
+        message: str,
+        success: bool,
+        queue: str,
+        *,
+        check_system: bool = False,
+        include_llm_message: bool = True
+    ):
+        """
+        Build a ResultObj-style message on the wire, mirroring your .NET usage.
+        check_system=True -> don't publish if obj.IsSystemLlm is True (matches your rule).
+        """
+        obj.ResultMessage = message
+        obj.ResultSuccess = success
+        if include_llm_message:
+            obj.LlmMessage = f"<Success>{message}</Success>" if success else f"<Error>{message}</Error>"
+
+        if check_system and obj.IsSystemLlm:
+            return
+
+        # You publish LLMServiceObj on "llmServiceMessage"/"llmSessionMessage" in .NET
+        # That lets the coordinator show the assistant text and statuses.
+        await self._pub.publish(queue, obj)
+
+    def _session_for(self, session_id: str) -> Optional[_Session]:
+        return self._sessions.get(session_id)
+
+    # ---------------------------- API methods ----------------------------
+
+    async def StartProcess(self, payload: Any):
+        llm = self._to_model(payload)
 
-    async def StartProcess(self, llmServiceObj: dict):
-        session_id = f"{llmServiceObj['RequestSessionId']}_{llmServiceObj['LLMRunnerType']}"
-        llmServiceObj["SessionId"] = session_id
+        # Construct Python-side session id like C#: RequestSessionId + "_" + LLMRunnerType
+        session_id = f"{llm.RequestSessionId}_{llm.LLMRunnerType}"
+        llm.SessionId = session_id
 
-        # wait ready (max ~120s like .NET)
+        # Wait ready (max 120s) exactly like the C# logic
         try:
             await asyncio.wait_for(self._ready.wait(), timeout=120)
         except asyncio.TimeoutError:
-            await self._set_result(llmServiceObj, "Timed out waiting for initialization.", False, "llmServiceMessage", True)
+            await self._emit_result(
+                llm, "Timed out waiting for initialization.", False, "llmServiceMessage", check_system=True
+            )
             return
 
-        sess = self._sessions.get(session_id)
-        is_runner_null = not sess or not sess.get("Runner")
+        sess = self._session_for(session_id)
+        is_runner_null = (sess is None) or (sess.Runner is None)
+        create_new = is_runner_null or (sess and sess.Runner and sess.Runner.IsStateFailed)
 
-        create_new = is_runner_null or sess["Runner"].IsStateFailed
         if create_new:
-            if sess and sess.get("Runner"):
+            # Remove previous runner if exists
+            if sess and sess.Runner:
                 try:
-                    await sess["Runner"].RemoveProcess(session_id)
-                except: pass
+                    await sess.Runner.RemoveProcess(session_id)
+                except Exception:
+                    pass
 
-            runner: ILLMRunner = await self._runner_factory(llmServiceObj)
+            # Create runner from factory
+            runner: ILLMRunner = await self._runner_factory(llm.model_dump())
             if not runner.IsEnabled:
-                await self._set_result(llmServiceObj, f"{llmServiceObj['LLMRunnerType']} {settings.SERVICE_ID} not started as it is disabled.", True, "llmServiceMessage")
+                await self._emit_result(
+                    llm,
+                    f"{llm.LLMRunnerType} {settings.SERVICE_ID} not started as it is disabled.",
+                    True,
+                    "llmServiceMessage",
+                )
                 return
 
-            await self._set_result(llmServiceObj, f"Starting {runner.Type} {settings.SERVICE_ID} Expert", True, "llmServiceMessage", True)
-            await runner.StartProcess(llmServiceObj)
+            await self._emit_result(
+                llm, f"Starting {runner.Type} {settings.SERVICE_ID} Expert", True, "llmServiceMessage", check_system=True
+            )
 
-            self._sessions[session_id] = {"Runner": runner, "FullSessionId": session_id}
-            if settings.SERVICE_ID == "monitor":
-                await self._set_result(llmServiceObj, f"Hi i'm {runner.Type} your Network Monitor Assistant. How can I help you.", True, "llmServiceMessage", True)
+            await runner.StartProcess(llm.model_dump())
 
-        await self._pub.publish("llmServiceStarted", llmServiceObj)
+            self._sessions[session_id] = _Session(Runner=runner, FullSessionId=session_id)
+
+            # Mirror your friendly greeting, gated by service id (you renamed yours to gradllm)
+            if settings.SERVICE_ID.lower() in {"monitor", "gradllm"}:
+                await self._emit_result(
+                    llm,
+                    f"Hi i'm {runner.Type} your {settings.SERVICE_ID} Assistant. How can I help you.",
+                    True,
+                    "llmServiceMessage",
+                    check_system=True,
+                )
+
+        # Notify "started"
+        await self._pub.publish("llmServiceStarted", llm)
+
+    async def RemoveSession(self, payload: Any):
+        llm = self._to_model(payload)
+        base = (llm.SessionId or "").split("_")[0]
+        targets = [k for k in list(self._sessions.keys()) if k.startswith(base + "_")]
 
-    async def RemoveSession(self, llmServiceObj: dict):
-        # Behaves like your RemoveAllSessionIdProcesses (prefix match)
-        base = llmServiceObj.get("SessionId","").split("_")[0]
-        targets = [k for k in self._sessions.keys() if k.startswith(base + "_")]
         msgs = []
         ok = True
         for sid in targets:
             s = self._sessions.get(sid)
-            if s and s.get("Runner"):
-                try:
-                    await s["Runner"].RemoveProcess(sid)
-                    s["Runner"] = None
-                    msgs.append(sid)
-                except Exception as e:
-                    ok = False
-                    msgs.append(f"Error {sid}: {e}")
+            if not s or not s.Runner:
+                continue
+            try:
+                await s.Runner.RemoveProcess(sid)
+                s.Runner = None
+                msgs.append(sid)
+            except Exception as e:
+                ok = False
+                msgs.append(f"Error {sid}: {e}")
+
         if ok:
-            await self._set_result(llmServiceObj, f"Success: Removed sessions for {' '.join(msgs)}", True, "llmSessionMessage", True)
+            await self._emit_result(
+                llm,
+                f"Success: Removed sessions for {' '.join(msgs) if msgs else '(none)'}",
+                True,
+                "llmSessionMessage",
+                check_system=True,
+            )
         else:
-            await self._set_result(llmServiceObj, " ".join(msgs), False, "llmServiceMessage")
-
-    async def StopRequest(self, llmServiceObj: dict):
-        sid = llmServiceObj.get("SessionId","")
-        s = self._sessions.get(sid)
-        if not s or not s.get("Runner"):
-            await self._set_result(llmServiceObj, f"Error: Runner missing for session {sid}.", False, "llmServiceMessage")
+            await self._emit_result(llm, " ".join(msgs), False, "llmServiceMessage")
+
+    async def StopRequest(self, payload: Any):
+        llm = self._to_model(payload)
+        sid = llm.SessionId or ""
+        s = self._session_for(sid)
+        if not s or not s.Runner:
+            await self._emit_result(llm, f"Error: Runner missing for session {sid}.", False, "llmServiceMessage")
             return
-        await s["Runner"].StopRequest(sid)
-        await self._set_result(llmServiceObj, f"Success {s['Runner'].Type} {settings.SERVICE_ID} Assistant output has been halted", True, "llmServiceMessage", True)
-
-    async def UserInput(self, llmServiceObj: dict):
-        sid = llmServiceObj.get("SessionId","")
-        s = self._sessions.get(sid)
-        if not s or not s.get("Runner"):
-            await self._set_result(llmServiceObj, f"Error: SessionId {sid} has no running process.", False, "llmServiceMessage")
+
+        await s.Runner.StopRequest(sid)
+        await self._emit_result(
+            llm,
+            f"Success {s.Runner.Type} {settings.SERVICE_ID} Assistant output has been halted",
+            True,
+            "llmServiceMessage",
+            check_system=True,
+        )
+
+    async def UserInput(self, payload: Any):
+        llm = self._to_model(payload)
+        sid = llm.SessionId or ""
+        s = self._session_for(sid)
+        if not s or not s.Runner:
+            await self._emit_result(llm, f"Error: SessionId {sid} has no running process.", False, "llmServiceMessage")
             return
-        r: ILLMRunner = s["Runner"]
+
+        r: ILLMRunner = s.Runner
         if r.IsStateStarting:
-            await self._set_result(llmServiceObj, "Please wait, the assistant is starting...", False, "llmServiceMessage")
+            await self._emit_result(llm, "Please wait, the assistant is starting...", False, "llmServiceMessage")
             return
         if r.IsStateFailed:
-            await self._set_result(llmServiceObj, "The Assistant is stopped. Try reloading.", False, "llmServiceMessage")
+            await self._emit_result(llm, "The Assistant is stopped. Try reloading.", False, "llmServiceMessage")
             return
-        await r.SendInputAndGetResponse(llmServiceObj)
-        # emitter side can push partials directly to queues if desired
 
-    async def QueryIndexResult(self, queryIndexRequest: dict):
-        # Adapted to your behavior: concatenate outputs, publish completion via internal coordinator if needed
+        # Let runner push partials itself if desired; we still return a small ack
+        await r.SendInputAndGetResponse(llm.model_dump())
+
+    async def QueryIndexResult(self, payload: Any):
+        """
+        Your .NET listener concatenates RAG outputs, sets ResultObj, and notifies the coordinator.
+        Here, we forward a service message containing the same info so the UI can reflect completion.
+        'payload' usually has: Success, Message, QueryResults: [{Output: "..."}]
+        """
         try:
-            rag_data = "\n".join([qr.get("Output","") for qr in (queryIndexRequest.get("QueryResults") or [])])
-            # You signal _queryCoordinator.CompleteQuery in .NET; here you may forward/publish result…
-            # Example: include rag data in a service message to the session
-            await self._pub.publish("llmServiceMessage", {
-                "ResultSuccess": queryIndexRequest.get("Success", False),
-                "ResultMessage": queryIndexRequest.get("Message",""),
-                "Data": rag_data,
-            })
+            data = payload if isinstance(payload, dict) else {}
+            outputs = data.get("QueryResults") or []
+            rag_data = "\n".join([x.get("Output", "") for x in outputs if isinstance(x, dict)])
+
+            # Shape compatible with your coordinator expectations
+            await self._pub.publish(
+                "llmServiceMessage",
+                ResultObj(
+                    Message=data.get("Message", ""),
+                    Success=bool(data.get("Success", False)),
+                    Data=rag_data,
+                )
+            )
         except Exception as e:
-            await self._pub.publish("llmServiceMessage", {"ResultSuccess": False, "ResultMessage": str(e)})
+            await self._pub.publish(
+                "llmServiceMessage",
+                ResultObj(Message=str(e), Success=False)
+            )
 
     async def GetFunctionRegistry(self, filtered: bool = False):
-        # Plug in your registry
-        data = {"FunctionCatalogJson": "{}", "Filtered": filtered}
-        await self._pub.publish("llmServiceMessage", {"ResultSuccess": True, "ResultMessage": f"Success : Got GetFunctionCatalogJson : {data}"})
+        """
+        Wire up to your real registry when ready.
+        For now, mimic your success message payload.
+        """
+        catalog = "{}"  # replace with real JSON
+        msg = f"Success : Got GetFunctionCatalogJson : {catalog}"
+        await self._pub.publish(
+            "llmServiceMessage",
+            ResultObj(Message=msg, Success=True)
+        )