A2A streaming using Dapr pub sub (#4)

* feat: A2A streaming works using Dapr pub-sub but things need to be cleaned up.

* fix: Corrected LaTeX math-mode display in Gradio chatbot by adding escaped square braces characters as LaTeX delimiters.

* chore: Added an environment variable to optionally specify a remote A2A MHQA endpoint.

* fix: Corrected parsing issues in the A2A CLI client for both streaming and non-streaming modes.
note: Read the A2A protocol documentation for details.

* chore: Reduced the number of messages the actor posts to the pub-sub queue.
fix: Corrected message statuses to _complete_ in the message history.

* chore: Added yfmcp MCP server for stock tickers.
chore: Minor UI modifications.

* fix: Corrected test condition for total number of MCP tools.

Author: anirbanbasu

.dapr/components/pubsub.yaml

@@ -10,3 +10,6 @@ spec:
     value: localhost:6379
   - name: redisPassword
     value: ""
+scopes:
+- dapr-srv
+- mhqa-a2a-srv

README.md

@@ -47,8 +47,10 @@ The following environment variables are also relevant but not essential, except
  - `BROWSER_STATE_SECRET`: This is the secret used by Gradio to encrypt the browser state data. The default value is `a2a_dapr_bstate_secret`.
  - `BROWSER_STATE_CHAT_HISTORIES`: This is the key in browser state used by Gradio to store the chat histories (local values). The default value is `a2a_dapr_chat_histories`.
  - `APP_DAPR_SVC_HOST` and `APP_DAPR_SVC_PORT`: The host and port at which Dapr actor service will listen on. These default to `127.0.0.1` and `32768`. Should you change these, you must change the corresponding information in `dapr.yaml`.
+ - `APP_DAPR_PUBSUB_STALE_MSG_SECS`: This specifies how old a message should be on the Dapr publish-subscribe topic queue before it will be considered too old, and dropped. The default value is 60 seconds.
  - `DAPR_PUBSUB_NAME`: The configured name of the publish-subscribe component at `.dapr/components/pubsub.yaml`. Change this environment variable only if you change the corresponding pub-sub component configuration.
  - `APP_A2A_SRV_HOST` and `APP_MHQA_A2A_SRV_PORT`: The host and port at which A2A endpoint will be available. These default to `127.0.0.1` and `32770`. Should you change these, you must change the corresponding information in `dapr.yaml`.
+ - `APP_MHQA_A2A_REMOTE_URL`: This environment variable can be used to specify the full remote URL including the protocol, i.e., `http` or `https` where the MHQA A2A endpoint is available. This is useful in a scenario where the web app is deployed on a machine that is different from where the MHQA A2A endpoint and Dapr service are. Default value is `None`.
 
 ## Usage
 

pyproject.toml

@@ -9,6 +9,7 @@ authors = [
 requires-python = ">=3.12.0"
 dependencies = [
     "a2a-sdk[http-server]>=0.3.6",
+    "anyio>=4.11.0",
     "dapr>=1.16.0",
     "dapr-ext-fastapi>=1.16.0",
     "environs>=14.3.0",
@@ -19,6 +20,7 @@ dependencies = [
     "llama-index-tools-mcp>=0.4.1",
     "ollama>=0.6.0",
     "typer>=0.19.1",
+    "yfmcp>=0.4.8",
 ]
 
 [project.scripts]

src/dqa/__init__.py

@@ -30,8 +30,12 @@ class ParsedEnvVars:
     MCP_CONFIG_FILE: str = env.str("MCP_CONFIG_FILE", default="conf/mcp.json")
     APP_DAPR_SVC_HOST: str = env.str("APP_DAPR_SVC_HOST", default="127.0.0.1")
     APP_DAPR_SVC_PORT: int = env.int("APP_DAPR_SVC_PORT", default=32768)
+    APP_DAPR_PUBSUB_STALE_MSG_SECS: int = env.int(
+        "APP_DAPR_PUBSUB_STALE_MSG_SECS", default=60
+    )
     APP_A2A_SRV_HOST: str = env.str("APP_A2A_SRV_HOST", default="127.0.0.1")
     APP_MHQA_A2A_SRV_PORT: int = env.int("APP_MHQA_A2A_SRV_PORT", default=32770)
+    APP_MHQA_A2A_REMOTE_URL: str = env.str("APP_MHQA_A2A_REMOTE_URL", default=None)
     APP_ECHO_A2A_SRV_PORT: int = env.int("APP_ECHO_A2A_SRV_PORT", default=32769)
     DAPR_PUBSUB_NAME: str = env.str("DAPR_PUBSUB_NAME", default="pubsub")
     MCP_SERVER_HOST: str = env.str("FASTMCP_HOST", default="localhost")
@@ -57,5 +61,5 @@ def __new__(cls: type["ParsedEnvVars"]) -> "ParsedEnvVars":
     level=ParsedEnvVars().APP_LOG_LEVEL,
     format="%(message)s",
     datefmt="[%X]",
-    handlers=[RichHandler()],
+    handlers=[RichHandler(show_time=True, show_level=True, show_path=True)],
 )

src/dqa/actor/__init__.py

@@ -1,8 +1,8 @@
-from enum import StrEnum, auto
+from enum import StrEnum
 
 
 class MHQAActorMethods(StrEnum):
-    Respond = auto()
-    GetChatHistory = auto()
-    ResetChatHistory = auto()
-    Cancel = auto()
+    Respond = "Respond"
+    GetChatHistory = "GetChatHistory"
+    ResetChatHistory = "ResetChatHistory"
+    Cancel = "Cancel"

src/dqa/actor/mhqa.py

@@ -11,6 +11,7 @@
 from llama_index.core.base.llms.types import ChatMessage, MessageRole
 from llama_index.core.tools.types import ToolOutput
 
+
 from llama_index.core.agent.workflow import (
     AgentOutput,
     ToolCall,
@@ -19,7 +20,6 @@
     AgentWorkflow,
     FunctionAgent,
 )
-
 from abc import abstractmethod
 import os
 from dapr.actor import Actor, ActorInterface, actormethod
@@ -28,7 +28,8 @@
 
 from dqa import ParsedEnvVars
 from dqa.actor import MHQAActorMethods
-from dqa.model.mhqa import MCPToolInvocation, MHQAResponse
+from dqa.actor.pubsub_topics import PubSubTopics
+from dqa.model.mhqa import MCPToolInvocation, MHQAResponse, MHQAResponseStatus
 
 
 logger = logging.getLogger(__name__)
@@ -221,6 +222,7 @@ async def respond(self, data: dict) -> dict:
         )
         full_response = ""
         tool_invocations: List[MCPToolInvocation] = []
+        pubsub_topic_name = f"{PubSubTopics.MHQA_RESPONSE}/{self.id}"
         with DaprClient() as dc:
             async for ev in wf_handler.stream_events():
                 if isinstance(ev, AgentStream):
@@ -260,11 +262,24 @@ async def respond(self, data: dict) -> dict:
                     agent_output=full_response,
                     tool_invocations=tool_invocations,
                 )
-                dc.publish_event(
-                    pubsub_name=ParsedEnvVars().DAPR_PUBSUB_NAME,
-                    topic_name=f"topic-{self.__class__.__name__}-{self.id}-respond",
-                    data=response.model_dump_json().encode(),
-                )
+                if (
+                    isinstance(ev, AgentStream)
+                    and ev.delta.strip() != ""
+                    and response
+                    and response.agent_output.strip() != ""
+                ):
+                    # logger.info(f"Publishing: {response.agent_output}")
+                    dc.publish_event(
+                        pubsub_name=ParsedEnvVars().DAPR_PUBSUB_NAME,
+                        topic_name=pubsub_topic_name,
+                        data=response.model_dump_json().encode(),
+                    )
+            response.status = MHQAResponseStatus.completed
+            dc.publish_event(
+                pubsub_name=ParsedEnvVars().DAPR_PUBSUB_NAME,
+                topic_name=pubsub_topic_name,
+                data=response.model_dump_json().encode(),
+            )
         memory_messages = await self.workflow_memory.aget_all()
         await self._state_manager.set_state(
             self._chat_memory_key,
@@ -293,6 +308,7 @@ async def get_chat_history(self) -> list:
                                 user_input=user_input,
                                 agent_output=agent_output,
                                 tool_invocations=tool_invocations,
+                                status=MHQAResponseStatus.completed,
                             )
                         )
                         tool_name = ""

src/dqa/actor/pubsub_topics.py

@@ -0,0 +1,2 @@
+class PubSubTopics:
+    MHQA_RESPONSE = "ps-topics/mhqa-response"

src/dqa/cli/a2a.py

@@ -73,7 +73,10 @@ def _initialize(self):
         self.echo_base_url = f"http://{a2a_asgi_host}:{echo_a2a_asgi_port}"
 
         mhqa_a2a_asgi_port = ParsedEnvVars().APP_MHQA_A2A_SRV_PORT
-        self.mhqa_base_url = f"http://{a2a_asgi_host}:{mhqa_a2a_asgi_port}"
+        self.mhqa_base_url = (
+            ParsedEnvVars().APP_MHQA_A2A_REMOTE_URL
+            or f"http://{a2a_asgi_host}:{mhqa_a2a_asgi_port}"
+        )
         logger.debug(f"Echo A2A base URL: {self.echo_base_url}")
         logger.debug(f"MHQA A2A base URL: {self.mhqa_base_url}")
 
@@ -277,8 +280,8 @@ async def _mhqa_chat(
             logger.info("Parsing streaming response from the A2A endpoint")
             full_message_content = ""
             async for response in streaming_response:
-                if isinstance(response, Message):
-                    full_message_content += get_message_text(response)
+                if response[0].status.message:
+                    full_message_content = get_message_text(response[0].status.message)
             validated_response = MHQAResponse.model_validate_json(full_message_content)
             return validated_response
 
@@ -324,8 +327,8 @@ async def _mhqa_get_history(
             logger.info("Parsing streaming response from the A2A endpoint")
             full_message_content = ""
             async for response in streaming_response:
-                if isinstance(response, Message):
-                    full_message_content += get_message_text(response)
+                if response[0].status.message:
+                    full_message_content = get_message_text(response[0].status.message)
             response_adapter = TypeAdapter(List[MHQAResponse])
             validated_response = response_adapter.validate_json(full_message_content)
             validated_response = validated_response[
@@ -374,8 +377,8 @@ async def _mhqa_delete_history(
             logger.info("Parsing streaming response from the A2A endpoint")
             full_message_content = ""
             async for response in streaming_response:
-                if isinstance(response, Message):
-                    full_message_content += get_message_text(response)
+                if response[0].status.message:
+                    full_message_content = get_message_text(response[0].status.message)
             return full_message_content
 
     async def run_mhqa_delete_history(
@@ -387,7 +390,9 @@ async def run_mhqa_delete_history(
             response = await self._mhqa_delete_history(
                 thread_id=thread_id,
             )
-            print(response)
+            print(
+                f"Deletion of thread '{thread_id}': {'successful' if response == 'true' else 'failed; maybe the thread does not exist?'}"
+            )
         except Exception as e:
             logger.error(f"Error in MHQA delete history. {e}")
         finally:
@@ -486,7 +491,7 @@ def mhqa_get_history(
 
 
 @app.command()
-async def mhqa_delete_history(
+def mhqa_delete_history(
     thread_id: str = typer.Option(
         help="A thread ID to identify your conversation.",
     ),

src/dqa/client/a2a_mixin.py

@@ -24,17 +24,17 @@ async def obtain_a2a_client(
         final_agent_card_to_use: AgentCard | None = None
 
         logger.info(
-            f"Attempting to fetch public agent card from: {base_url}{AGENT_CARD_WELL_KNOWN_PATH}"
+            f"Fetching A2A agent card from: {base_url}{AGENT_CARD_WELL_KNOWN_PATH}"
         )
         _public_card = (
             await resolver.get_agent_card()
         )  # Fetches from default public path
         logger.info("Successfully fetched public agent card.")
-        logger.info(_public_card.model_dump_json(indent=2, exclude_none=True))
+        logger.debug(_public_card.model_dump_json(indent=2, exclude_none=True))
         final_agent_card_to_use = _public_card
 
         client = ClientFactory(
             config=ClientConfig(streaming=True, polling=True, httpx_client=httpx_client)
         ).create(card=final_agent_card_to_use)
-        logger.info("A2A client initialised.")
+        logger.debug("A2A client initialised.")
         return client, final_agent_card_to_use