HELP NEEDED: Handling terminal state for TaskUpdater

[waitasecant]

UPDATE: Removing the await updater.complete() and break statement resolved he issue.

How to handle terminal state nicely? This doesn't break the interactive loop in client.py possibly because the streaming data is already sent before the error occurs. Thanks for your time. 🙏

agent_executor.py

class CurrencyAgentExecutor(AgentExecutor):
    """AgentExecutor wrapper for Agents"""

    def __init__(
        self,
        session_timestamp: str,
        mcp_tools: Optional[List[Any]] = None,
    ):
        """Initialize with MCP tools passed from main application"""
        # Configure logging
        self.logger = config_log(session_timestamp, "AGENT-EXECUTOR")

        self.agent = CurrencyAgent(
            mcp_tools=mcp_tools, session_timestamp=session_timestamp
        )
        tool_count = len(mcp_tools) if mcp_tools else 0
        self.logger.info(f"AgentExecutor initialized with {tool_count} MCP tools")

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        query = context.get_user_input()
        task = context.current_task
        if not task:
            if not context.message:
                raise ServerError(error=InvalidParamsError())
            task = new_task(context.message)
            await event_queue.enqueue_event(task)
        updater = TaskUpdater(event_queue, task.id, task.context_id)

        try:
            # Get direct response from the agent
            self.logger.info(f"Processing query: {query}")
            complete = False
            async for event in self.agent.invoke(query, task.context_id):
                self.logger.info(f"Event received: {event}")
                if "data" in event:
                    msg = event["data"]
                else:
                    msg = ""
                if "result" in event:
                    result = event["result"]
                    if hasattr(result, "stop_reason"):
                        if "end_turn" == result.stop_reason:
                            complete = True
                            self.logger.info(f"End of response")

                            await updater.update_status(
                                TaskState.completed,
                                new_agent_text_message(
                                    msg,
                                    task.context_id,
                                    task.id,
                                ),
                            )
                            await updater.add_artifact([Part(root=TextPart(text=msg))])
                            self.logger.info("Final artifact added")
                            await updater.complete()
                            self.logger.info("Task completed successfully")
                            break
                else:
                    await updater.update_status(
                        TaskState.working,
                        new_agent_text_message(
                            msg,
                            task.context_id,
                            task.id,
                        ),
                    )
                    await updater.add_artifact([Part(root=TextPart(text=msg))])

                self.logger.info(f"Received text: {msg}")

        except Exception as e:
            self.logger.error(f"An error occurred while processing the query: {e}")
            raise ServerError(error=InternalError()) from e

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise ServerError(error=UnsupportedOperationError())

agent.py

    async def invoke(self, query: str, context_id: str):
        """Process a query and return the response"""
        # Initialize conversation for this context if not exists
        if context_id not in self.conversations:
            self.conversations[context_id] = Agent(
                model=self.model,
                tools=self.tools,
                system_prompt=self.SYSTEM_INSTRUCTION,
            )

        agent = self.conversations[context_id]

        try:
            import os
            from contextlib import redirect_stdout, redirect_stderr

            # Suppress stdout/stderr during agent execution in server
            with open(os.devnull, "w") as devnull:
                with redirect_stdout(devnull), redirect_stderr(devnull):
                    async for event in agent.stream_async(query):
                        yield event

        except Exception as e:
            self.logger.error(f"Error in query processing: {e}")
            yield {}

logs

2025-08-14 15:38:56,397 - AGENT-EXECUTOR - INFO - Event received: {'event': {'contentBlockDelta': {'delta': {'text': ' a list of supported currencies?'}, 'contentBlockIndex': 0}}}
2025-08-14 15:38:56,398 - AGENT-EXECUTOR - INFO - Received text: 
2025-08-14 15:38:56,398 - AGENT-EXECUTOR - INFO - Event received: {'data': ' a list of supported currencies?', 'delta': {'text': ' a list of supported currencies?'}, 'agent': <strands.agent.agent.Agent object at 0x0000023BBF0D1A90>, 'event_loop_cycle_id': UUID('f276231f-e0f2-4c14-8fbb-a85f9a1a1c25'), 'request_state': {}, 'event_loop_cycle_trace': <strands.telemetry.metrics.Trace object at 0x0000023BBF0D2F90>, 'event_loop_cycle_span': NonRecordingSpan(SpanContext(trace_id=0x00000000000000000000000000000000, span_id=0x0000000000000000, trace_flags=0x00, trace_state=[], is_remote=False))}
2025-08-14 15:38:56,399 - AGENT-EXECUTOR - INFO - Received text:  a list of supported currencies?
2025-08-14 15:38:56,404 - AGENT-EXECUTOR - INFO - Event received: {'event': {'contentBlockStop': {'contentBlockIndex': 0}}}
2025-08-14 15:38:56,405 - AGENT-EXECUTOR - INFO - Received text: 
2025-08-14 15:38:56,405 - AGENT-EXECUTOR - INFO - Event received: {'event': {'messageStop': {'stopReason': 'end_turn'}}}
2025-08-14 15:38:56,406 - AGENT-EXECUTOR - INFO - Received text: 
2025-08-14 15:38:56,406 - AGENT-EXECUTOR - INFO - Event received: {'event': {'metadata': {'usage': {'inputTokens': 1027, 'outputTokens': 63, 'totalTokens': 1090}, 'metrics': {'latencyMs': 2345}}}}
2025-08-14 15:38:56,406 - AGENT-EXECUTOR - INFO - Received text: 
2025-08-14 15:38:56,408 - AGENT-EXECUTOR - INFO - Event received: {'message': {'role': 'assistant', 'content': [{'text': "Hello! Welcome to the currency conversion assistant. I'm here to help you with currency exchange rates and conversions. How can I assist you today? Would you like to know the current exchange rate between two currencies, convert an amount from one currency to another, or see a list of supported currencies?"}]}}
2025-08-14 15:38:56,408 - AGENT-EXECUTOR - INFO - Received text: 
2025-08-14 15:38:56,409 - AGENT-EXECUTOR - INFO - Event received: {'result': AgentResult(stop_reason='end_turn', message={'role': 'assistant', 'content': [{'text': "Hello! Welcome to the currency conversion assistant. I'm here to help you with currency exchange rates and conversions. How can I assist you today? Would you like to know the current exchange rate between two currencies, convert an amount from one currency to another, or see a list of supported currencies?"}]}, metrics=EventLoopMetrics(cycle_count=1, tool_metrics={}, cycle_durations=[3.470747947692871], traces=[<strands.telemetry.metrics.Trace object at 0x0000023BBF0D2F90>], accumulated_usage={'inputTokens': 1027, 'outputTokens': 63, 'totalTokens': 1090}, accumulated_metrics={'latencyMs': 2345}), state={})}
2025-08-14 15:38:56,409 - AGENT-EXECUTOR - INFO - End of response
2025-08-14 15:38:56,409 - AGENT-EXECUTOR - INFO - Final artifact added
2025-08-14 15:38:56,409 - AGENT-EXECUTOR - ERROR - An error occurred while processing the query: Task 1f3ffe17-b91d-484e-8124-972497eba2d2 is already in a terminal state.
2025-08-14 15:38:56,410 - opentelemetry.context - ERROR - Failed to detach context
Traceback (most recent call last):
  File "C:\Users\Hi617\Code\A2A-MCP\venv\Lib\site-packages\opentelemetry\trace\__init__.py", line 589, in use_span
    yield span
  File "C:\Users\Hi617\Code\A2A-MCP\venv\Lib\site-packages\strands\agent\agent.py", line 515, in stream_async
    yield {"result": result}
GeneratorExit

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Users\Hi617\Code\A2A-MCP\venv\Lib\site-packages\opentelemetry\context\__init__.py", line 155, in detach
    _RUNTIME_CONTEXT.detach(token)
    ~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^
  File "C:\Users\Hi617\Code\A2A-MCP\venv\Lib\site-packages\opentelemetry\context\contextvars_context.py", line 53, in detach
    self._current_context.reset(token)
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^
ValueError: <Token var=<ContextVar name='current_context' default={} at 0x0000023BBBBA0450> at 0x0000023BBF0FD500> was created in a different Context