Merge pull request #192 from pdettori/fix/await-task-updater-187

fix(a2a): await async TaskUpdater methods in currency converter

Author: pdettori

a2a/a2a_currency_converter/app/agent.py

@@ -9,11 +9,16 @@
 from langchain_core.tools import tool
 from langchain_openai import ChatOpenAI
 from langgraph.checkpoint.memory import MemorySaver
+from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 from langgraph.prebuilt import create_react_agent
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings
 
-memory = MemorySaver()
+memory = MemorySaver(
+    serde=JsonPlusSerializer(
+        allowed_msgpack_modules=[("app.agent", "ResponseFormat")],
+    ),
+)
 
 
 class Configuration(BaseSettings):

a2a/a2a_currency_converter/app/agent_executor.py

@@ -60,7 +60,7 @@ async def execute(
 
                 if not is_task_complete and not require_user_input:
                     logger.info(f"Updating status for non-input task: {task.id}")
-                    updater.update_status(
+                    await updater.update_status(
                         TaskState.working,
                         new_agent_text_message(
                             item["content"],
@@ -70,7 +70,7 @@ async def execute(
                     )
                 elif require_user_input:
                     logger.info(f"Updating status for input task: {task.id}")
-                    updater.update_status(
+                    await updater.update_status(
                         TaskState.input_required,
                         new_agent_text_message(
                             item["content"],
@@ -82,11 +82,11 @@ async def execute(
                     break
                 else:
                     logger.info("Adding artifact for item")
-                    updater.add_artifact(
+                    await updater.add_artifact(
                         [Part(root=TextPart(text=item["content"]))],
                         name="conversion_result",
                     )
-                    updater.complete()
+                    await updater.complete()
                     break
 
         except InternalServerError as e:
@@ -100,7 +100,7 @@ async def execute(
 """
             logger.error(msg=msg)
             logger.error(msg=f"Raw InternalServerError: {e}")
-            updater.update_status(
+            await updater.update_status(
                 TaskState.input_required,
                 new_agent_text_message(
                     msg,
@@ -123,7 +123,7 @@ async def execute(
 The key should match your OpenAI key."""
             logger.error(msg=msg)
             logger.error(msg=f"Raw AuthenticationError {e}")
-            updater.update_status(
+            await updater.update_status(
                 TaskState.input_required,
                 new_agent_text_message(
                     msg,
@@ -136,7 +136,7 @@ async def execute(
         except Exception as e:
             logger.error(f"An error occurred while streaming the response: {e}")
             logger.info(msg=f"The error is a {type(e)}")
-            updater.update_status(
+            await updater.update_status(
                 TaskState.input_required,
                 new_agent_text_message(
                     # We don't show the error to the user, as it may have credentials

tests/a2a/test_currency_converter.py

@@ -15,6 +15,8 @@
     "langgraph",
     "langgraph.checkpoint",
     "langgraph.checkpoint.memory",
+    "langgraph.checkpoint.serde",
+    "langgraph.checkpoint.serde.jsonplus",
     "langgraph.prebuilt",
 ]:
     sys.modules.setdefault(mod, MagicMock())