Get Artifacts From RemoteA2aAgent into Orchestrator Agent session

[codemonkey-df]

Im looking for answer how:

Get Artifacts From RemoteA2aAgent into Orchestrator Agent session

---

I run Orchestrator Agent which has subagent set to RemoteA2aAgent. And this Remote Agent may return some artifacts like images. Im looking for answer how to save this artifacts into orchestrator agent session

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: To transfer artifacts from a RemoteA2aAgent to an OrchestratorAgent, use context.save_artifact() in the remote agent's logic and context.load_artifact() in the orchestrator's logic. Both agents share the same artifact storage when running in the same session, so artifacts saved by one are immediately available to the other.

---

Hello! That's a great question about handling artifacts in a multi-agent setup.

The key principle to understand is that all agents operating within the same session share the same InvocationContext. [1, 2] This context provides access to shared services, including the ArtifactService, which manages binary data like images and files. [3, 4]

When your OrchestratorAgent invokes a RemoteA2aAgent as a sub-agent, they are both part of that same session. This means an artifact saved by a tool within the RemoteA2aAgent is registered with the session's ArtifactService and becomes immediately available to the OrchestratorAgent (or any other agent in that session).

Here’s a conceptual workflow for how you can achieve this:

1. In your Remote Agent's Logic (or a tool it calls):

   • First, generate or receive your artifact data (e.g., the bytes of an image).
   • Create a Part object from this data.
   • Use context.save_artifact() to store it. It's a good practice to then store the artifact's name in the shared session state so the orchestrator knows what to look for. [2]

   from google.generativeai import types
   
   # This code would be in a tool called by your RemoteA2aAgent
   def generate_and_save_image(query: str, context) -> str:
       image_bytes = ... # Your logic to generate the image
       image_part = types.Part(
           inline_data=types.Blob(mime_type="image/png", data=image_bytes)
       )
   
       artifact_name = "generated_image.png"
       context.save_artifact(artifact_name, image_part)
   
       # Save the name to the shared state for the orchestrator to find
       context.state["temp:generated_image_name"] = artifact_name
   
       return f"Successfully generated and saved artifact: {artifact_name}"

2. In your Orchestrator Agent's Logic:

   • After the RemoteA2aAgent has finished its turn, you can access the shared state to get the name of the artifact.
   • Use context.load_artifact() with that name to retrieve the Part object and access its data.

   # This code would be in a tool called by your OrchestratorAgent
   # after the RemoteA2aAgent has run.
   def process_remote_artifact(query: str, context) -> str:
       artifact_name = context.state.pop("temp:generated_image_name", None)
   
       if not artifact_name:
           return "No artifact name found in state."
   
       # Load the artifact from the shared ArtifactService
       image_part = context.load_artifact(artifact_name)
       image_bytes = image_part.inline_data.data
   
       # Now you can process or save the image bytes in the orchestrator's environment
       with open(f"local_copy_of_{artifact_name}", "wb") as f:
           f.write(image_bytes)
   
       return f"Successfully loaded and saved artifact {artifact_name} locally."

This works because the Runner that executes your session is initialized with an ArtifactService (like InMemoryArtifactService or GcsArtifactService), and this single service instance is used for the entire duration of the session. [3]

[1] https://google.github.io/adk-docs/get-started/about/
[2] https://google.github.io/adk-docs/context/
[3] https://google.github.io/adk-docs/artifacts/
[4] https://github.com/google/adk-python/blob/main/contributing/adk_project_overview_and_architecture.md

[codemonkey-df]

Im talking about example implementation:

from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent

prime_agent = RemoteA2aAgent(
    name="prime_agent",
    description="Agent that handles checking if numbers are prime.",
    agent_card=(
        f"http://localhost:8001/a2a/check_prime_agent{AGENT_CARD_WELL_KNOWN_PATH}"
    ),
)

from google.adk.agents.llm_agent import Agent
from google.genai import types

root_agent = Agent(
    model="gemini-2.0-flash",
    name="root_agent",
    instruction="""
      <You are a helpful assistant that can roll dice and check if numbers are prime.
      You delegate rolling dice tasks to the roll_agent and prime checking tasks to the prime_agent.
      Follow these steps:
      1. If the user asks to roll a die, delegate to the roll_agent.
      2. If the user asks to check primes, delegate to the prime_agent.
      3. If the user asks to roll a die and then check if the result is prime, call roll_agent first, then pass the result to prime_agent.
      Always clarify the results before proceeding.>
    """,
    global_instruction=(
        "You are DicePrimeBot, ready to roll dice and check prime numbers."
    ),
    sub_agents=[roll_agent, prime_agent],
    tools=[example_tool],
    generate_content_config=types.GenerateContentConfig(
        safety_settings=[
            types.SafetySetting(  # avoid false alarm about rolling dice.
                category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=types.HarmBlockThreshold.OFF,
            ),
        ]
    ),
)