Grafi 49/pub sub refactoring (#74)

remove sync flow
change everything to async
add serial workflow
add event queue interface

Author: GuanyiLi-Craig

docs/docs/getting-started/quickstart.md

@@ -85,9 +85,12 @@ Create a file named `react_agent_app.py` and create a build-in react-agent:
 
 ```python
 # react_agent_app.py
+import asyncio
+
 from grafi.agents.react_agent import create_react_agent
 
-def main():
+
+async def main():
     print("ReAct Agent Chat Interface")
     print("Type your questions and press Enter. Type '/bye' to exit.")
     print("-" * 50)
@@ -97,7 +100,7 @@ def main():
     while True:
         user_input = input("\nYou: ").strip()
 
-        if user_input.lower() == '/bye':
+        if user_input.lower() == "/bye":
             print("Goodbye!")
             break
 
@@ -106,20 +109,22 @@ def main():
 
         try:
             # Get synchronized response from agent
-            output = react_agent.run(user_input)
+            output = await react_agent.run(user_input)
             print(f"\nAgent: {output}")
 
         except Exception as e:
             print(f"Error: {e}")
 
+
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())
 ```
 
-And finally export an `OPENAI_API_KEY` key as an environment variable:
+And finally export an `OPENAI_API_KEY` and `TAVILY_API_KEY` key as an environment variable:
 
 ```bash
 export OPENAI_API_KEY="sk-proj-******"
+export TAVILY_API_KEY="tvly-******"
 ```
 
 ---

docs/docs/guide/configuring-event-store.md

@@ -130,20 +130,25 @@ event_store = container.event_store
 Graphite has a built in react agent that can be used out of the box for your needs, we will use it for the simple use case of passing in input and retrieving the output from OpenAI.
 
 ```python
+import asyncio
 from grafi.agents.react_agent import create_react_agent
 
-react_agent = create_react_agent()
+async def run_agent():
 
-result = react_agent.run(user_input, invoke_context)
+    react_agent = create_react_agent()
 
-print("Output from React Agent:", result)
+    result = await react_agent.run(user_input, invoke_context)
 
+    print("Output from React Agent:", result)
 
-events = event_store.get_conversation_events(conversation_id)
 
-print(f"Events for conversation {conversation_id}:")
+    events = event_store.get_conversation_events(conversation_id)
 
-print(f"Events: {events} ")
+    print(f"Events for conversation {conversation_id}:")
+
+    print(f"Events: {events} ")
+
+asyncio.run(run_agent())
 
 ```
 
@@ -167,8 +172,8 @@ import uuid
 from grafi.agents.react_agent import create_react_agent
 from grafi.common.containers.container import container
 from grafi.common.event_stores.event_store_postgres import EventStorePostgres
-from grafi.common.models.invoke_context import InvokeContext
-from grafi.common.models.message import Message
+from grafi.models.invoke_context import InvokeContext
+from grafi.models.message import Message
 
 postgres_event_store = EventStorePostgres(
     db_url="postgresql://postgres:postgres@localhost:5432/grafi_test_db",
@@ -202,21 +207,22 @@ invoke_context = InvokeContext(
     assistant_request_id=assistant_request_id,
 )
 
-message = Message(role="user", content=user_input)
+async def run_agent():
+
+    react_agent = create_react_agent()
 
+    result = await react_agent.run(user_input, invoke_context)
 
-react_agent = create_react_agent()
+    print("Output from React Agent:", result)
 
-result = react_agent.run(user_input, invoke_context)
 
+    events = event_store.get_conversation_events(conversation_id)
 
-for output_message in result:
-    print("Output message:", output_message)
+    print(f"Events for conversation {conversation_id}:")
 
-events = event_store.get_conversation_events(conversation_id)
+    print(f"Events: {events} ")
 
-print(f"Events for conversation {conversation_id}:")
-print(f"Events: {events} ")
+asyncio.run(run_agent())
 
 ```
 
@@ -246,10 +252,10 @@ print(f"Events: {events} ")
 2025-07-12 17:38:55.566 | WARNING  | grafi.tools.function_calls.function_call_tool:__init_subclass__:86 - GoogleSearchTool: no method decorated with @llm_function found.
 2025-07-12 17:38:55.895 | DEBUG    | grafi.common.instrumentations.tracing:is_local_endpoint_available:30 - Endpoint check failed: [Errno 111] Connection refused
 2025-07-12 17:38:55.895 | DEBUG    | grafi.common.instrumentations.tracing:setup_tracing:117 - OTLP endpoint is not available. Using InMemorySpanExporter.
-2025-07-12 17:38:55.905 | INFO     | grafi.common.topics.topic:publish_data:64 - [agent_input_topic] Message published with event_id: 1da4f45008264dc98fd63dc154dcaa6e
+2025-07-12 17:38:55.905 | INFO     | grafi.topics.topic:publish_data:64 - [agent_input_topic] Message published with event_id: 1da4f45008264dc98fd63dc154dcaa6e
 2025-07-12 17:38:55.912 | DEBUG    | grafi.nodes.node:invoke:49 - Executing Node with inputs: [ConsumeFromTopicEvent(event_id='8dd13f159fbc40b29ca030501f4d2859', event_version='1.0', invoke_context=InvokeContext(conversation_id='9051491ca7a84b71a3f3e9d790b79e4f', invoke_id='6e56da25ca824bf4a8d454a731b6f336', assistant_request_id='0e0317d25c9b40e4a76a130bbbb7bc43', user_id='9e396be8f8c442bc9269a31a9b3f69a2'), event_type=<EventType.CONSUME_FROM_TOPIC: 'ConsumeFromTopic'>, timestamp=datetime.datetime(2025, 7, 12, 16, 38, 55, 909090, tzinfo=datetime.timezone.utc), name='agent_input_topic', offset=0, data=[Message(name=None, message_id='ae2638fa5f3047fbbfb43dbd1545e6d2', timestamp=1752338335885634346, content='What is the capital of the United Kingdom', refusal=None, annotations=None, audio=None, role='user', tool_call_id=None, tools=None, function_call=None, tool_calls=None, is_streaming=False)], consumer_name='OpenAIInputNode', consumer_type='OpenAIInputNode')]
-2025-07-12 17:38:57.400 | INFO     | grafi.common.topics.topic:publish_data:69 - [function_call_topic] Message NOT published (condition not met)
-2025-07-12 17:38:57.400 | INFO     | grafi.common.topics.output_topic:publish_data:91 - [agent_output_topic] Message published with event_id: d83d3339fbd94d6e983fc51965177891
+2025-07-12 17:38:57.400 | INFO     | grafi.topics.topic:publish_data:69 - [function_call_topic] Message NOT published (condition not met)
+2025-07-12 17:38:57.400 | INFO     | grafi.topics.output_topic:publish_data:91 - [agent_output_topic] Message published with event_id: d83d3339fbd94d6e983fc51965177891
 <span style="color:#FF4689">Output from React Agent:</span>  The capital of the United Kingdom is London.
 <span style="color:#FF4689">Events for conversation: </span> 9051491ca7a84b71a3f3e9d790b79e4f:
 <span style="color:#FF4689">Events: </span>[AssistantInvokeEvent(event_id='7d97059ad3404c74be582ed4d01e365e', event_version='1.0', invoke_context=InvokeContext(conversation_id='9051491ca7a84b71a3f3e9d790b79e4f', invoke_id='6e56da25ca824bf4a8d454a731b6f336', assistant_request_id='0e0317d25c9b40e4a76a130bbbb7bc43', user_id='9e396be8f8c442bc9269a31a9b3f69a2'), event_type=<EventType.ASSISTANT_INVOKE: 'AssistantInvoke'>, timestamp=datetime.datetime(2025, 7, 12, 16, 38, 55, 885710), assistant_id='de51912ecc6c44ce803a0cd5eb358ba5', assistant_name='ReActAgent', assistant_type='ReActAgent', input_data=[Message(name=None, message_id='ae2638fa5f3047fbbfb43dbd1545e6d2', timestamp=1752338335885634346, content='What is the capital of the United Kingdom', refusal=None, annotations=No```</code></pre></div>

docs/docs/guide/connecting-to-an-mcp-server.md

@@ -176,10 +176,10 @@ from pydantic import Field
 
 from grafi.assistants.assistant import Assistant
 from grafi.assistants.assistant_base import AssistantBaseBuilder
-from grafi.common.topics.input_topic import InputTopic
-from grafi.common.topics.output_topic import OutputTopic
-from grafi.common.topics.subscription_builder import SubscriptionBuilder
-from grafi.common.topics.topic import Topic
+from grafi.topics.input_topic import InputTopic
+from grafi.topics.output_topic import OutputTopic
+from grafi.topics.subscription_builder import SubscriptionBuilder
+from grafi.topics.topic import Topic
 from grafi.nodes.node import Node
 from grafi.tools.function_calls.function_call_tool import FunctionCallTool
 from grafi.tools.llms.impl.openai_tool import OpenAITool
@@ -337,7 +337,7 @@ class StockAssistantBuilder(
 ```
 
 
-Graphtie is natively asychrnous, but you can chose to run syncrhous coroutines as well. For this case we are making a fully asynchronous workflow by overrding the  `async def a_run()` method of the `Assistant` class. In order to run this create a `main.py` that will instantiate the assistant and execute it asynchrnously.
+Graphtie is natively asychrnous, but you can chose to run syncrhous coroutines as well. For this case we are making a fully asynchronous workflow by overrding the  `async def run()` method of the `Assistant` class. In order to run this create a `main.py` that will instantiate the assistant and execute it asynchrnously.
 
 ```python
 #main.py
@@ -347,9 +347,10 @@ import uuid
 from typing import Dict
 
 from grafi.common.containers.container import container
-from grafi.common.models.invoke_context import InvokeContext
-from grafi.common.models.mcp_connections import StreamableHttpConnection
-from grafi.common.models.message import Message
+from grafi.common.events.topic_events.publish_to_topic_event import PublishToTopicEvent
+from grafi.models.invoke_context import InvokeContext
+from grafi.models.mcp_connections import StreamableHttpConnection
+from grafi.models.message import Message
 from grafi.tools.function_calls.impl.mcp_tool import MCPTool
 
 from assistant import StockAssistant
@@ -372,7 +373,7 @@ async def create_assistant():
         .model(model)
         .api_key(api_key)
         .function_call_llm_system_message(function_call_llm_system_message)
-        .function_tool(await MCPTool.builder().connections(mcp_config).a_build())
+        .function_tool(await MCPTool.builder().connections(mcp_config).build())
         .summary_llm_system_message(
             "You are a helpful assistant that provides a summary of the company overview. return all output as json formatted string")
         .build()
@@ -392,7 +393,11 @@ async def main():
     question = "What is the overview of the company Tesla?"
     input_data = [Message(role="user", content=question)]
 
-    async for response in assistant.a_invoke(invoke_context, input_data):
+    publish_event = PublishToTopicEvent(
+        invoke_context=execution_context, data=input_messages
+    )
+
+    async for response in assistant.invoke(publish_event):
         print("Assistant output:")
         for output in response:
             print(output.content)
@@ -524,11 +529,11 @@ Topics can have conditions attached to them to trigger them, in the case of `Inp
 
 
 
-When you call the assistant's `a_invoke()` method from `main.py`
+When you call the assistant's `invoke()` method from `main.py`
 
 ```python
 
-async for output in assistant.a_invoke(invoke_context, input_data):
+async for output in assistant.invoke(invoke_context, input_data):
 ```
 
 The `input_data` gets published on the assistants `InputTopic` during workflow initialization. This is done [within the framework](https://github.com/binome-dev/graphite/blob/0b6b666b6f0e122ac970f437ee24f5d87c30a81a/grafi/workflows/impl/event_driven_workflow.py#L681-L687) but we are outlying it here so you can follow the flow of data.

docs/docs/guide/creating-a-simple-workflow.md

@@ -43,8 +43,8 @@ The main function orchestrates the entire workflow. We start by defining a sampl
 
 ```python linenums="9"
 import uuid
-from grafi.common.models.message import Message
-from grafi.common.models.invoke_context import InvokeContext
+from grafi.models.message import Message
+from grafi.models.invoke_context import InvokeContext
 
 def main():
     user_input = "What is the capital of the United Kingdom"
@@ -85,8 +85,8 @@ The node subscribes to the `agent_input_topic` (the root topic in Graphite), use
 ```python linenums="27"
 from grafi.nodes.node import Node
 from grafi.tools.llms.impl.openai_tool import OpenAITool
-from grafi.common.topics.input_topic import InputTopic
-from grafi.common.topics.output_topic import OutputTopic
+from grafi.topics.input_topic import InputTopic
+from grafi.topics.output_topic import OutputTopic
 
 
 agent_input_topic = InputTopic(name="agent_input_topic")
@@ -131,13 +131,12 @@ workflow = (
 With the `EventDrivenWorkflow` object created, we can invoke it by passing our `invoke_context` and a `List[Message]`. The workflow will execute and return the results, which we can then print. Save this complete code as `main.py`.
 
 ```python linenums="54"
-result = workflow.invoke(
+async for result in workflow.invoke(
     invoke_context,
     [message]
-)
-
-for output_message in result:
-    print("Output message:", output_message.content)
+):
+    for output_message in result:
+        print("Output message:", output_message.content)
 ```
 
 ### 6. Entry Point

docs/docs/guide/getting-started-with-assistants.md

@@ -129,8 +129,8 @@ Implement the builder pattern for fluent configuration of the assistant. This pi
 
 ```python
 from grafi.nodes.node import Node
-from grafi.common.topics.input_topic import InputTopic
-from grafi.common.topics.output_topic import OutputTopic
+from grafi.topics.input_topic import InputTopic
+from grafi.topics.output_topic import OutputTopic
 from grafi.tools.llms.impl.openai_tool import OpenAITool
 from grafi.workflows.impl.event_driven_workflow import EventDrivenWorkflow
 
@@ -188,15 +188,16 @@ Prepare input data and context for workflow execution:
 - Returns both the input data and context
 
 ```python
-from grafi.common.models.invoke_context import InvokeContext
+from grafi.common.events.topic_events.publish_to_topic_event import PublishToTopicEvent
+from grafi.models.invoke_context import InvokeContext
 from typing import Optional
-from grafi.common.models.message import Message
+from grafi.models.message import Message
 
 class FinanceAssistant(Assistant):
 
     ...
 
-    def get_input(self, question: str, invoke_context: Optional[InvokeContext] = None) -> tuple[list[Message], InvokeContext]:
+    def get_input(self, question: str, invoke_context: Optional[InvokeContext] = None) -> PublishToTopicEvent:
         """Prepare input data and invoke context."""
         if invoke_context is None:
             logger.debug("Creating new InvokeContext with default conversation id for FinanceAssistant")
@@ -214,7 +215,9 @@ class FinanceAssistant(Assistant):
             )
         ]
 
-        return input_data, invoke_context
+        return PublishToTopicEvent(
+            invoke_context=execution_context, data=input_messages
+        )
 ```
 
 
@@ -227,22 +230,22 @@ This function is not part of the framework, but rather a helper function used to
 class FinanceAssistant(Assistant):
 
     ...
-    def run(self, question: str, invoke_context: Optional[InvokeContext] = None) -> str:
+    async def run(self, question: str, invoke_context: Optional[InvokeContext] = None) -> str:
         """Run the assistant with a question and return the response."""
         # Call helper function get_input()
-        input_data, invoke_context = self.get_input(question, invoke_context)
+        input_event= self.get_input(question, invoke_context)
         # This is the line that invokes the workflow
-        output = super().invoke(invoke_context, input_data)
-
-        # Handle different content types
-        if output and len(output) > 0:
-            content = output[0].content
-            if isinstance(content, str):
-                return content
-            elif content is not None:
-                return str(content)
-
-        return "No response generated"
+        response_str = ""
+        async for output in super().invoke(input_event):
+            # Handle different content types
+            if output and len(output) > 0:
+                content = output.data[0].content
+                if isinstance(content, str):
+                    response_str += content
+                elif content is not None:
+                    response_str += str(content)
+
+        return response_str
 ```
 
 Main execution method that:
@@ -258,6 +261,8 @@ Now that we have created the class for the assistance, we have to instantiate it
 
 ```python
 # main.py
+import asyncio
+
 def main():
     system_message = os.getenv("OPENAI_SYSTEM_MESSAGE")
     api_key = os.getenv("OPENAI_API_KEY")
@@ -274,12 +279,12 @@ def main():
 
     """Main function to run the assistant."""
     user_input = "What are the key factors to consider when choosing between a 401(k) and a Roth IRA?"
-    result = assistant.run(user_input)
+    result = await assistant.run(user_input)
     print("Output message:", result)
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())
 
 ```
 
@@ -319,12 +324,12 @@ def main():
 
     """Main function to run the assistant."""
     user_input = "What are the key factors to consider when choosing between a 401(k) and a Roth IRA?"
-    result = assistant.run(user_input)
+    result = await assistant.run(user_input)
     print("Output message:", result)
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())
 ```