HEX
Server: Apache
System: Linux scp1.abinfocom.com 5.4.0-216-generic #236-Ubuntu SMP Fri Apr 11 19:53:21 UTC 2025 x86_64
User: confeduphaar (1010)
PHP: 8.1.33
Disabled: exec,passthru,shell_exec,system
Upload Files
File: //proc/self/root/lib/mysqlsh/lib/python3.8/site-packages/oci/addons/adk/run/response.py
# coding: utf-8
# Copyright (c) 2016, 2025, Oracle and/or its affiliates.  All rights reserved.
# This software is dual-licensed to you under the Universal Permissive License (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose either license.

"""
ADK run response
"""

import json
from typing import Any, Dict

from pydantic import BaseModel, Field, computed_field

from oci.addons.adk.logger import default_logger as logger


class RunResponse(BaseModel):
    """Response from the agent run containing the final output."""

    data: Dict[str, Any] = Field(..., description="Raw response from the agent")
    session_id: str = Field(..., description="Session ID used for the response")

    @computed_field  # type: ignore[prop-decorator]
    @property
    def final_output(self) -> str | None:
        """Direct access to the agent message text as output"""
        return self._get_nested_text()

    @computed_field(deprecated="Use 'final_output' instead")  # type: ignore[prop-decorator]
    @property
    def output(self) -> str | None:
        """Direct access to the agent message text as output"""
        return self.final_output

    def _get_nested_text(self) -> str | None:
        """Safely extract text from nested data structure."""
        try:
            return self.data["message"]["content"]["text"]
        except (KeyError, TypeError):
            return None

    def pretty_print(self, print_data: bool = False) -> None:
        """Pretty print this run response."""
        if self.final_output is None:
            logger.warning("NO TEXT AVAILABLE IN RESPONSE")

        message_text = f"agent text message:\n[bold green]{self.final_output}[/bold green]"

        if print_data:
            message_text += f"\n\ndata:\n[bold magenta]{json.dumps(self.data, indent=4)}[/bold magenta]"  # noqa: E501

        logger.print(
            message_text,
            title="Agent run response",
            border_style="blue",
        )