Artifact stored as text is not retrieved as text #3157

import os
import asyncio
from google.adk.agents import Agent
from google.adk.sessions import InMemorySessionService, Session
from google.adk.artifacts import GcsArtifactService
from google.adk.runners import Runner
from google.adk.tools import BaseTool
from google.adk.tools.tool_context import ToolContext
from google.genai import types
import logging
import json


logging.basicConfig(level=logging.ERROR)

os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"
os.environ["GOOGLE_CLOUD_PROJECT"] = "your-favourite-project"
os.environ["GOOGLE_CLOUD_LOCATION"] = "your-region"
ARTIFACTS_BUCKET = "your-favourite-bucket"

AGENT_MODEL = "gemini-2.5-flash-lite"
APP_NAME = "weather_tutorial_app"
USER_ID = "user_1"
SESSION_ID = "session_001" # Using a fixed ID for simplicity


def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city.

    Args:
        city (str): The name of the city (e.g., "New York", "London", "Tokyo").

    Returns:
        dict: A dictionary containing the weather information.
              Includes a 'status' key ('success' or 'error').
              If 'success', includes a 'report' key with weather details.
              If 'error', includes an 'error_message' key.
    """
    city_normalized = city.lower().replace(" ", "")
    mock_weather_db = {
        "newyork": {"status": "success", "report": "The weather in New York is sunny with a temperature of 25°C."},
        "london": {"status": "success", "report": "It's cloudy in London with a temperature of 15°C."},
        "tokyo": {"status": "success", "report": "Tokyo is experiencing light rain and a temperature of 18°C."},
    }
    if city_normalized in mock_weather_db:
        return mock_weather_db[city_normalized]
    else:
        return {"status": "error", "error_message": f"Sorry, I don't have weather information for '{city}'."}

async def collect_artifacts(tool:BaseTool, args, tool_context:ToolContext, tool_response):
    if tool.name == 'transfer_to_agent':
        return
    await tool_context.save_artifact('weather', types.Part.from_text(text=json.dumps(tool_response)))

root_agent = Agent(
    name="weather_agent",
    model=AGENT_MODEL,
    description="The weather agent. Answers questions about the weather in certain places.",
    instruction="You are the Weather Agent. Your primary responsibility is to provide weather information. "
                "Use the 'get_weather' tool for specific weather requests (e.g., 'weather in London'). "
                "For anything else, respond appropriately or state you cannot handle it.",
    tools=[get_weather],
    after_tool_callback=collect_artifacts
)

async def run_agent(runner: Runner, session: Session, content):
    print(" ================================= ")
    final_response_text = ''

    def from_content(content: types.Part):
        if not content:
            return '<None>'
        else:
            if content.text:
                return content.text.replace('\n', '\\n')
            else:
                return '<data structure>'
            
    async for event in runner.run_async(user_id=session.user_id, session_id=session.id, new_message=content):
        if event and event.content and event.content.parts:
            for part in event.content.parts:
                if part.thought:
                    event_type = 'Thought'
                elif part.text:
                    event_type = 'Text'
                elif part.function_call:
                    event_type = 'Function Call'
                elif part.function_response:
                    event_type = 'Function Response'
                else:
                    event_type = 'Other event'
                print(f"-->  [Event] Author: {event.author}, Type: {event_type}, Final: {event.is_final_response()}, Actions: {event.actions is not None}, Content: {from_content(part)}")
                print()
                if event_type == 'Text' and event.is_final_response() and part.text:
                    final_response_text += f'{event.author}: {part.text} '
        if event.is_final_response() and event.actions and event.actions.escalate:
            final_response_text = f"Agent escalated: {event.error_message or 'No specific message.'}"

    print(f"<<< Agent Response: {final_response_text if len(final_response_text) else 'Could not identify final response.'}")
    if runner.artifact_service:
        if artifact := await runner.artifact_service.load_artifact(app_name=session.app_name, user_id=session.user_id, session_id=session.id, filename='weather'):
            print(f'Weather artifact text contains: {artifact.text}')
            print(f'inline data: {artifact.inline_data}')

async def main():
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=root_agent, app_name=session.app_name, session_service=session_service, artifact_service=GcsArtifactService(bucket_name=ARTIFACTS_BUCKET))
    content = types.Content(role='user', parts=[types.Part(text='Tell me about the weather in New York')])
    await run_agent(runner, session, content)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"An error occurred: {e}")