Added RunErrorDetails object for MaxTurnsExceeded exception

src/agents/__init__.py

@@ -44,7 +44,7 @@
 from .models.openai_chatcompletions import OpenAIChatCompletionsModel
 from .models.openai_provider import OpenAIProvider
 from .models.openai_responses import OpenAIResponsesModel
-from .result import RunResult, RunResultStreaming
+from .result import RunErrorDetails, RunResult, RunResultStreaming
 from .run import RunConfig, Runner
 from .run_context import RunContextWrapper, TContext
 from .stream_events import (
@@ -204,6 +204,7 @@ def enable_verbose_stdout_logging():
  "AgentHooks",
  "RunContextWrapper",
  "TContext",
+ "RunErrorDetails",
  "RunResult",
  "RunResultStreaming",
  "RunConfig",

src/agents/exceptions.py

@@ -1,7 +1,10 @@
+from __future__ import annotations
+
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
  from .guardrail import InputGuardrailResult, OutputGuardrailResult
+ from .result import RunErrorDetails
 
 
 class AgentsException(Exception):
@@ -12,9 +15,11 @@ class MaxTurnsExceeded(AgentsException):
  """Exception raised when the maximum number of turns is exceeded."""
 
  message: str
+ run_error_details: RunErrorDetails | None
 
- def __init__(self, message: str):
+ def __init__(self, message: str, run_error_details: RunErrorDetails | None = None):
  self.message = message
+ self.run_error_details = run_error_details
 
 
 class ModelBehaviorError(AgentsException):
@@ -40,10 +45,10 @@ def __init__(self, message: str):
 class InputGuardrailTripwireTriggered(AgentsException):
  """Exception raised when a guardrail tripwire is triggered."""
 
- guardrail_result: "InputGuardrailResult"
+ guardrail_result: InputGuardrailResult
  """The result data of the guardrail that was triggered."""
 
- def __init__(self, guardrail_result: "InputGuardrailResult"):
+ def __init__(self, guardrail_result: InputGuardrailResult):
  self.guardrail_result = guardrail_result
  super().__init__(
  f"Guardrail {guardrail_result.guardrail.__class__.__name__} triggered tripwire"
@@ -53,10 +58,10 @@ def __init__(self, guardrail_result: "InputGuardrailResult"):
 class OutputGuardrailTripwireTriggered(AgentsException):
  """Exception raised when a guardrail tripwire is triggered."""
 
- guardrail_result: "OutputGuardrailResult"
+ guardrail_result: OutputGuardrailResult
  """The result data of the guardrail that was triggered."""
 
- def __init__(self, guardrail_result: "OutputGuardrailResult"):
+ def __init__(self, guardrail_result: OutputGuardrailResult):
  self.guardrail_result = guardrail_result
  super().__init__(
  f"Guardrail {guardrail_result.guardrail.__class__.__name__} triggered tripwire"

src/agents/result.py

@@ -18,7 +18,11 @@
 from .run_context import RunContextWrapper
 from .stream_events import StreamEvent
 from .tracing import Trace
-from .util._pretty_print import pretty_print_result, pretty_print_run_result_streaming
+from .util._pretty_print import (
+ pretty_print_result,
+ pretty_print_run_error_details,
+ pretty_print_run_result_streaming,
+)
 
 if TYPE_CHECKING:
  from ._run_impl import QueueCompleteSentinel
@@ -244,3 +248,35 @@ def _cleanup_tasks(self):
 
  def __str__(self) -> str:
  return pretty_print_run_result_streaming(self)
+
+
+@dataclass
+class RunErrorDetails:
+ input: str | list[TResponseInputItem]
+ """The original input items i.e. the items before run() was called. This may be a mutated
+ version of the input, if there are handoff input filters that mutate the input.
+ """
+
+ new_items: list[RunItem]
+ """The new items generated during the agent run. These include things like new messages, tool
+ calls and their outputs, etc.
+ """
+
+ raw_responses: list[ModelResponse]
+ """The raw LLM responses generated by the model during the agent run."""
+
+ input_guardrail_results: list[InputGuardrailResult]
+ """Guardrail results for the input messages."""
+
+ context_wrapper: RunContextWrapper[Any]
+ """The context wrapper for the agent run."""
+
+ _last_agent: Agent[Any]
+
+ @property
+ def last_agent(self) -> Agent[Any]:
+ """The last agent that was run."""
+ return self._last_agent
+
+ def __str__(self) -> str:
+ return pretty_print_run_error_details(self)

src/agents/run.py

@@ -1,3 +1,4 @@
+
 from __future__ import annotations
 
 import asyncio
@@ -35,7 +36,7 @@
 from .model_settings import ModelSettings
 from .models.interface import Model, ModelProvider
 from .models.multi_provider import MultiProvider
-from .result import RunResult, RunResultStreaming
+from .result import RunErrorDetails, RunResult, RunResultStreaming
 from .run_context import RunContextWrapper, TContext
 from .stream_events import AgentUpdatedStreamEvent, RawResponsesStreamEvent
 from .tool import Tool
@@ -208,7 +209,9 @@ async def run(
  data={"max_turns": max_turns},
  ),
  )
- raise MaxTurnsExceeded(f"Max turns ({max_turns}) exceeded")
+ raise MaxTurnsExceeded(
+ f"Max turns ({max_turns}) exceeded"
+ )
 
  logger.debug(
  f"Running agent {current_agent.name} (turn {current_turn})",
@@ -283,6 +286,23 @@ async def run(
  raise AgentsException(
  f"Unknown next step type: {type(turn_result.next_step)}"
  )
+ except Exception as e:
+ run_error_details = RunErrorDetails(
+ input=original_input,
+ new_items=generated_items,
+ raw_responses=model_responses,
+ input_guardrail_results=input_guardrail_results,
+ context_wrapper=context_wrapper,
+ _last_agent=current_agent
+ )
+ # Re-raise with the error details
+ if isinstance(e, MaxTurnsExceeded):
+ raise MaxTurnsExceeded(
+ f"Max turns ({max_turns}) exceeded",
+ run_error_details
+ ) from e
+ else:
+ raise
  finally:
  if current_span:
  current_span.finish(reset_current=True)

src/agents/util/_pretty_print.py

@@ -3,7 +3,7 @@
 from pydantic import BaseModel
 
 if TYPE_CHECKING:
- from ..result import RunResult, RunResultBase, RunResultStreaming
+ from ..result import RunErrorDetails, RunResult, RunResultBase, RunResultStreaming
 
 
 def _indent(text: str, indent_level: int) -> str:
@@ -37,6 +37,15 @@ def pretty_print_result(result: "RunResult") -> str:
 
  return output
 
+def pretty_print_run_error_details(result: "RunErrorDetails") -> str:
+ output = "RunErrorDetails:"
+ output += f'\n- Last agent: Agent(name="{result.last_agent.name}", ...)'
+ output += f"\n- {len(result.new_items)} new item(s)"
+ output += f"\n- {len(result.raw_responses)} raw response(s)"
+ output += f"\n- {len(result.input_guardrail_results)} input guardrail result(s)"
+ output += "\n(See `RunErrorDetails` for more details)"
+
+ return output
 
 def pretty_print_run_result_streaming(result: "RunResultStreaming") -> str:
  output = "RunResultStreaming:"