feat(sdk): add error extension

Signed-off-by: Radek Ježek <radek.jezek@ibm.com>

Author: jezekra1

agents/chat/src/chat/agent.py

@@ -20,6 +20,9 @@
     BaseExtensionSpec,
     CitationExtensionServer,
     CitationExtensionSpec,
+    ErrorExtensionParams,
+    ErrorExtensionServer,
+    ErrorExtensionSpec,
     TrajectoryExtensionServer,
     TrajectoryExtensionSpec,
     LLMServiceExtensionServer,
@@ -89,8 +92,7 @@
                 description="Fetches summaries and information from Wikipedia articles.",
             ),
             AgentDetailTool(
-                name="Weather Information (OpenMeteo)",
-                description="Provides real-time weather updates and forecasts.",
+                name="Weather Information (OpenMeteo)", description="Provides real-time weather updates and forecasts."
             ),
             AgentDetailTool(
                 name="Web Search (DuckDuckGo)",
@@ -158,7 +160,8 @@ async def chat(
         LLMServiceExtensionServer,
         LLMServiceExtensionSpec.single_demand(suggested=("openai:gpt-4o", "ollama:granite3.3:8b")),
     ],
-    _: Annotated[PlatformApiExtensionServer, PlatformApiExtensionSpec()],
+    _e: Annotated[ErrorExtensionServer, ErrorExtensionSpec(ErrorExtensionParams(include_stacktrace=True))],
+    _p: Annotated[PlatformApiExtensionServer, PlatformApiExtensionSpec()],
 ):
     """Agent with memory and access to web search, Wikipedia, and weather."""
     await context.store(input)

apps/agentstack-sdk-py/examples/error_agent.py

@@ -0,0 +1,40 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+from typing import Annotated
+
+from a2a.types import Message
+
+from agentstack_sdk.a2a.extensions.ui.error import (
+    ErrorExtensionParams,
+    ErrorExtensionServer,
+    ErrorExtensionSpec,
+)
+from agentstack_sdk.server import Server
+
+server = Server()
+
+
+@server.agent()
+async def error_agent(
+    input: Message,
+    error_ext: Annotated[
+        ErrorExtensionServer,
+        ErrorExtensionSpec(params=ErrorExtensionParams(include_stacktrace=True)),
+    ],
+):
+    """Agent that demonstrates error handling using the ErrorExtension."""
+    yield "I am about to fail..."
+
+    # Intentionally raise an error to demonstrate the extension
+    # The server wrapper will catch this and use the injected error_ext to format the message
+    raise ValueError("This is a simulated error to demonstrate the ErrorExtension.")
+
+
+def run():
+    server.run(host=os.getenv("HOST", "127.0.0.1"), port=int(os.getenv("PORT", 10000)))
+
+
+if __name__ == "__main__":
+    run()

apps/agentstack-sdk-py/src/agentstack_sdk/a2a/extensions/ui/__init__.py

@@ -3,6 +3,7 @@
 
 from .agent_detail import *
 from .citation import *
+from .error import *
 from .form_request import *
 from .settings import *
 from .trajectory import *

apps/agentstack-sdk-py/src/agentstack_sdk/a2a/extensions/ui/error.py

@@ -0,0 +1,153 @@
+# Copyright 2025 © BeeAI a Series of LF Projects, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import traceback
+from types import NoneType
+from typing import Any
+
+import pydantic
+
+from agentstack_sdk.a2a.extensions.base import (
+    BaseExtensionClient,
+    BaseExtensionServer,
+    BaseExtensionSpec,
+)
+from agentstack_sdk.a2a.types import AgentMessage, Metadata
+
+
+class Error(pydantic.BaseModel):
+    """
+    Represents error information for displaying exceptions in the UI.
+
+    This extension helps display errors in a user-friendly way with:
+    - A clear error title (exception type)
+    - A descriptive error message
+    - Optional context dictionary for additional error details
+
+    Visually, this may appear as an error card in the UI.
+
+    Properties:
+    - title: Title of the error (typically the exception class name).
+    - message: The error message describing what went wrong.
+    - context: Optional dictionary with additional context about the error.
+    """
+
+    title: str
+    message: str
+    context: dict[str, Any] | None = None
+
+
+class ErrorMetadata(pydantic.BaseModel):
+    """
+    Metadata containing a list of errors and an optional stack trace.
+
+    Properties:
+    - errors: List of error objects describing what went wrong.
+    - stack_trace: Optional formatted stack trace for debugging (single trace for all errors).
+    """
+
+    errors: list[Error] = pydantic.Field(default_factory=list)
+    stack_trace: str | None = None
+
+
+class ErrorExtensionParams(pydantic.BaseModel):
+    """
+    Configuration parameters for the error extension.
+
+    Properties:
+    - include_stacktrace: Whether to include stack traces in error messages (default: False).
+    """
+
+    include_stacktrace: bool = False
+
+
+class ErrorExtensionSpec(BaseExtensionSpec[ErrorExtensionParams]):
+    URI: str = "https://a2a-extensions.agentstack.beeai.dev/ui/error/v1"
+
+
+def _format_stacktrace(exc: BaseException, include_cause: bool = True) -> str:
+    """Format exception with full traceback including nested causes."""
+    return "".join(traceback.format_exception(type(exc), exc, exc.__traceback__, chain=include_cause))
+
+
+def _extract_errors(exc: BaseException) -> list[Error]:
+    """
+    Extract error information from an exception, handling:
+    - BaseExceptionGroup (returns multiple errors)
+    - FrameworkError from beeai_framework (uses .explain() method)
+    - Nested exceptions via __cause__ chain
+    """
+    # Handle BaseExceptionGroup by recursively extracting errors from each exception
+    if isinstance(exc, BaseExceptionGroup):
+        return [error for sub_exc in exc.exceptions for error in _extract_errors(sub_exc)]
+
+    # Try to handle FrameworkError if beeai_framework is available
+    try:
+        from beeai_framework.errors import FrameworkError
+
+        if isinstance(exc, FrameworkError):
+            # FrameworkError has special .explain() method and context
+            return [Error(title=exc.name(), message=exc.explain(), context=exc.context or None)]
+    except ImportError:
+        # beeai_framework not installed, continue with standard handling
+        pass
+
+    return [
+        Error(
+            title=type(exc).__name__,
+            message=str(exc),
+            context=None,
+        )
+    ]
+
+
+class ErrorExtensionServer(BaseExtensionServer[ErrorExtensionSpec, NoneType]):
+    def error_metadata(self, error: BaseException) -> Metadata[str, Any]:
+        """
+        Create metadata for an error.
+
+        Args:
+            error: The exception to convert to metadata
+
+        Returns:
+            Metadata dictionary with error information
+        """
+        errors = _extract_errors(error)
+        stack_trace = _format_stacktrace(error) if self.spec.params.include_stacktrace else None
+        return Metadata({self.spec.URI: ErrorMetadata(errors=errors, stack_trace=stack_trace).model_dump(mode="json")})
+
+    def message(
+        self,
+        error: BaseException,
+    ) -> AgentMessage:
+        """
+        Create an AgentMessage with error metadata and serialized text representation.
+
+        Args:
+            error: The exception to include in the message
+
+        Returns:
+            AgentMessage with error metadata and markdown-formatted text
+        """
+        metadata = self.error_metadata(error)
+        error_metadata = ErrorMetadata.model_validate(metadata[self.spec.URI])
+
+        # Serialize to markdown for display
+        text_lines = []
+        for err in error_metadata.errors:
+            text_lines.append(f"**{err.title}**: {err.message}")
+            if err.context:
+                text_lines.append(f"\n*Context*: `{err.context}`")
+
+        text = "\n\n".join(text_lines)
+
+        # Add stack trace at the end if present
+        if error_metadata.stack_trace:
+            text += f"\n\n```\n{error_metadata.stack_trace}\n```"
+
+        return AgentMessage(text=text, metadata=metadata)
+
+
+class ErrorExtensionClient(BaseExtensionClient[ErrorExtensionSpec, ErrorMetadata]): ...

apps/agentstack-sdk-py/src/agentstack_sdk/server/agent.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import inspect
-import json
 from asyncio import CancelledError
 from collections.abc import AsyncGenerator, AsyncIterator, Callable, Generator
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
@@ -37,14 +36,14 @@
 )
 
 from agentstack_sdk.a2a.extensions.ui.agent_detail import AgentDetail, AgentDetailExtensionSpec
+from agentstack_sdk.a2a.extensions.ui.error import ErrorExtensionParams, ErrorExtensionServer, ErrorExtensionSpec
 from agentstack_sdk.a2a.types import ArtifactChunk, Metadata, RunYield, RunYieldResume
-from agentstack_sdk.server.constants import _IMPLICIT_DEPENDENCY_PREFIX
+from agentstack_sdk.server.constants import _IMPLICIT_DEPENDENCY_PREFIX, DEFAULT_ERROR_EXTENSION
 from agentstack_sdk.server.context import RunContext
 from agentstack_sdk.server.dependencies import extract_dependencies
 from agentstack_sdk.server.store.context_store import ContextStore
 from agentstack_sdk.server.utils import cancel_task, close_queue
 from agentstack_sdk.util.logging import logger
-from agentstack_sdk.util.utils import extract_messages
 
 AgentFunction: TypeAlias = Callable[[], AsyncGenerator[RunYield, RunYieldResume]]
 AgentFunctionFactory: TypeAlias = Callable[
@@ -124,9 +123,14 @@ def agent_factory(context_store: ContextStore):
             resolved_name = name or fn.__name__
             resolved_description = description or fn.__doc__ or ""
 
+            # Check if user has provided an ErrorExtensionServer, if not add default
+            has_error_extension = any(isinstance(ext, ErrorExtensionServer) for ext in sdk_extensions)
+            error_extension_spec = ErrorExtensionSpec(ErrorExtensionParams()) if not has_error_extension else None
+
             capabilities.extensions = [
                 *(capabilities.extensions or []),
                 *(AgentDetailExtensionSpec(detail).to_agent_card_extensions()),
+                *(error_extension_spec.to_agent_card_extensions() if error_extension_spec else []),
                 *(e_card for ext in sdk_extensions for e_card in ext.spec.to_agent_card_extensions()),
             ]
 
@@ -232,6 +236,11 @@ async def agent_executor_lifespan(
                             depends(message, context, dependency_args)
                         )
 
+                    context._error_extension = next(
+                        (ext for ext in sdk_extensions if isinstance(ext, ErrorExtensionServer)),
+                        DEFAULT_ERROR_EXTENSION,
+                    )
+
                     context._store = await context_store.create(
                         context_id=request_context.context_id,
                         initialized_dependencies=list(dependency_args.values()),
@@ -339,8 +348,9 @@ def with_context(message: Message | None = None) -> Message | None:
                 deep=True, update={"context_id": task_updater.context_id, "task_id": task_updater.task_id}
             )
 
+        run_context: RunContext | None = None
         try:
-            async with self._agent_executor_span(task_updater, context, context_store) as (execute_fn, _run_context):
+            async with self._agent_executor_span(task_updater, context, context_store) as (execute_fn, run_context):
                 agent_generator_fn = execute_fn()
 
                 await task_updater.start_work()
@@ -455,8 +465,11 @@ def with_context(message: Message | None = None) -> Message | None:
             await task_updater.cancel()
         except Exception as ex:
             logger.error("Error when executing agent", exc_info=ex)
-            msg = json.dumps(extract_messages(ex), indent=2)
-            await task_updater.failed(task_updater.new_agent_message(parts=[Part(root=TextPart(text=msg))]))
+            # Use the configured error extension from context, or create default with stacktrace enabled
+            error_extension = run_context._error_extension if run_context else None
+            error_extension = error_extension if error_extension is not None else DEFAULT_ERROR_EXTENSION
+            error_msg = error_extension.message(ex)
+            await task_updater.failed(error_msg)
         finally:  # cleanup
             await cancel_task(cancellation_task)
             is_cancelling = bool(current_task.cancelling())

apps/agentstack-sdk-py/src/agentstack_sdk/server/constants.py

@@ -3,6 +3,10 @@
 
 from typing import Final
 
+from agentstack_sdk.a2a.extensions.ui.error import ErrorExtensionParams, ErrorExtensionServer, ErrorExtensionSpec
+
 _IMPLICIT_DEPENDENCY_PREFIX: Final = "___server_dep"
 
+DEFAULT_ERROR_EXTENSION: Final = ErrorExtensionServer(ErrorExtensionSpec(ErrorExtensionParams()))
+
 __all__ = []

apps/agentstack-sdk-py/src/agentstack_sdk/server/context.py

@@ -3,6 +3,7 @@
 
 
 from collections.abc import AsyncIterator
+from typing import TYPE_CHECKING
 
 import janus
 from a2a.server.context import ServerCallContext
@@ -13,6 +14,9 @@
 from agentstack_sdk.a2a.types import RunYield, RunYieldResume
 from agentstack_sdk.server.store.context_store import ContextStoreInstance
 
+if TYPE_CHECKING:
+    from agentstack_sdk.a2a.extensions.ui.error import ErrorExtensionServer
+
 
 class RunContext(BaseModel, arbitrary_types_allowed=True):
     configuration: MessageSendConfiguration | None = None
@@ -26,6 +30,7 @@ class RunContext(BaseModel, arbitrary_types_allowed=True):
     _store: ContextStoreInstance | None = PrivateAttr(None)
     _yield_queue: janus.Queue[RunYield] = PrivateAttr(default_factory=janus.Queue)
     _yield_resume_queue: janus.Queue[RunYieldResume] = PrivateAttr(default_factory=janus.Queue)
+    _error_extension: "ErrorExtensionServer | None" = PrivateAttr(None)
 
     async def store(self, data: Message | Artifact):
         if not self._store: