Merge pull request #138 from esnible/currency-converter-logging

Feature: Currency converter logging and A2A error reporting

Author: esnible

a2a/a2a_currency_converter/app/__main__.py

@@ -37,10 +37,7 @@ class MissingAPIKeyError(Exception):
 def main(host, port):
     """Starts the Currency Agent server."""
     try:
-        if not os.getenv('OPENAI_API_KEY'):
-            raise MissingAPIKeyError(
-                'OPENAI_API_KEY environment variable not set.'
-            )
+        # We don't check OPENAI_API_KEY here.  We want the agent pod to run even if OPENAI_API_KEY isn't defined.
 
         capabilities = AgentCapabilities(streaming=True, pushNotifications=True)
         skill = AgentSkill(

a2a/a2a_currency_converter/app/agent.py

@@ -1,3 +1,7 @@
+"""Currency conversion logic for A2A example"""
+
+import os
+
 from collections.abc import AsyncIterable
 from typing import Any, Literal
 
@@ -10,16 +14,19 @@
 from langgraph.prebuilt import create_react_agent
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings
-import os
 from langchain_openai import ChatOpenAI
 
 
 memory = MemorySaver()
 
 class Configuration(BaseSettings):
+    """The configuration of the Agent"""
+
     llm_model: str = "gpt-4o"
     llm_api_base: str = ""
-    llm_api_key: str = os.getenv("OPENAI_API_KEY")
+    # We don't want the pod to crash without a valid key.
+    # Report authentication error to A2A user instead.
+    llm_api_key: str = os.getenv("OPENAI_API_KEY", "Failed to load env var OPENAI_API_KEY")
 
 
 config = Configuration()

a2a/a2a_currency_converter/app/agent_executor.py

@@ -1,4 +1,13 @@
+"""
+Example of the business logic of an A2A agent for currency conversion.
+"""
+
 import logging
+import os
+
+from app.agent import CurrencyAgent
+
+from openai import AuthenticationError, InternalServerError
 
 from a2a.server.agent_execution import AgentExecutor, RequestContext
 from a2a.server.events import EventQueue
@@ -18,9 +27,6 @@
 )
 from a2a.utils.errors import ServerError
 
-from app.agent import CurrencyAgent
-
-
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -38,12 +44,14 @@ async def execute(
     ) -> None:
         error = self._validate_request(context)
         if error:
+            logger.warning(f'Invalid agent executor request: {context}')
             raise ServerError(error=InvalidParamsError())
 
         query = context.get_user_input()
         task = context.current_task
         if not task:
             task = new_task(context.message)
+            logger.info(f'Created task for message : {context.message}')
             event_queue.enqueue_event(task)
         updater = TaskUpdater(event_queue, task.id, task.contextId)
         try:
@@ -52,6 +60,7 @@ async def execute(
                 require_user_input = item['require_user_input']
 
                 if not is_task_complete and not require_user_input:
+                    logger.info(f'Updating status for non-input task: {task.id}')
                     updater.update_status(
                         TaskState.working,
                         new_agent_text_message(
@@ -61,6 +70,7 @@ async def execute(
                         ),
                     )
                 elif require_user_input:
+                    logger.info(f'Updating status for input task: {task.id}')
                     updater.update_status(
                         TaskState.input_required,
                         new_agent_text_message(
@@ -72,21 +82,78 @@ async def execute(
                     )
                     break
                 else:
+                    logger.info('Adding artifact for item')
                     updater.add_artifact(
                         [Part(root=TextPart(text=item['content']))],
                         name='conversion_result',
                     )
                     updater.complete()
                     break
 
+        except InternalServerError as e:
+            msg=f"""CurrencyAgentExecutor reports an InternalServerError error.
+
+This can happen if the agent's LLM_API_BASE environment variable does not point to an OpenAI server.
+
+LLM_API_BASE is {os.getenv("LLM_API_BASE", "undefined")}
+
+Use `kubectl -n <namespace> logs deployment/<agent-name>` for details.
+"""
+            logger.error(msg=msg)
+            logger.error(msg=f"Raw InternalServerError: {e}")
+            updater.update_status(
+                TaskState.input_required,
+                new_agent_text_message(
+                    msg,
+                    task.contextId,
+                    task.id,
+                ),
+                final=True,
+            )
+
+        except AuthenticationError as e:
+            msg=f"""CurrencyAgentExecutor reports an authentication error.
+
+When importing this agent into Kagenti, expand Environment Variables and Add Variable,
+or import https://github.com/kagenti/agent-examples/blob/main/a2a/a2a_currency_converter/.env.openai
+
+Use `kubectl -n <namespace> logs deployment/<agent-name>` for details.
+
+Also check
+`kubectl -n <namespace> get secret openai-secret -o jsonpath="{'{'}.data.apikey{'}'}" | base64 -d`
+The key should match your OpenAI key."""
+            logger.error(msg=msg)
+            logger.error(msg=f"Raw AuthenticationError {e}")
+            updater.update_status(
+                TaskState.input_required,
+                new_agent_text_message(
+                    msg,
+                    task.contextId,
+                    task.id,
+                ),
+                final=True,
+            )
+
         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(
+                TaskState.input_required,
+                new_agent_text_message(
+                    # We don't show the error to the user, as it may have credentials
+                    """Internal error on the agent.
+                    Use `kubectl -n <namespace> logs deployment/<agent-name>` for details""",
+                    task.contextId,
+                    task.id,
+                ),
+                final=True,
+            )
             raise ServerError(error=InternalError()) from e
 
-    def _validate_request(self, context: RequestContext) -> bool:
+    def _validate_request(self, _: RequestContext) -> bool:
         return False
 
     async def cancel(
-        self, request: RequestContext, event_queue: EventQueue
+        self, _: RequestContext, event_queue: EventQueue
     ) -> Task | None:
         raise ServerError(error=UnsupportedOperationError())