/agents-build

multi-agent

Build AgentCore systems where agents delegate work to other agents.

When to use

• You want an orchestrator agent to delegate complex tasks to a specialist
• You're building a system where agents have different roles and capabilities
• You want agents to discover and communicate with each other via the A2A standard
• You want multiple agents to share the same memory

Input

$ARGUMENTS is optional:

/multi-agent              # interactive — asks which pattern you need
/multi-agent a2a          # A2A protocol setup
/multi-agent direct       # direct invocation pattern
/multi-agent memory       # shared memory across agents

Choosing a pattern

Step 1: Deploy the specialist agent

The specialist is a standard AgentCore agent. Deploy it normally:

agentcore create --name SpecialistAgent --defaults
# ... add your specialist logic to app/SpecialistAgent/main.py ...
agentcore deploy -y

Get the specialist's runtime ARN after deploy:

agentcore status --runtime SpecialistAgent --json | jq -r '.runtimes[0].arn'

Step 2: Add the specialist as a tool in the orchestrator

The orchestrator calls the specialist via bedrock-agentcore:InvokeAgentRuntime. Add this tool to your orchestrator's agent code:

import os
import json
import boto3
from bedrock_agentcore.runtime import BedrockAgentCoreApp

app = BedrockAgentCoreApp()

# Set this env var in your orchestrator's deployment config
SPECIALIST_ARN = os.getenv("SPECIALIST_AGENT_ARN")
REGION = os.getenv("AWS_REGION", "us-east-1")

def call_specialist(prompt: str, session_id: str = None) -> str:
    """
    Call the specialist agent and return its response.
    The specialist runs in its own isolated session.
    """
    client = boto3.client("bedrock-agentcore", region_name=REGION)

    kwargs = {
        "agentRuntimeArn": SPECIALIST_ARN,
        "qualifier": "DEFAULT",  # or a specific version number to pin
        "payload": json.dumps({"prompt": prompt}).encode(),
    }
    if session_id:
        kwargs["runtimeSessionId"] = session_id

    response = client.invoke_agent_runtime(**kwargs)
    # response["response"] is a StreamingBody — read, then parse JSON
    body = response["response"].read()
    result = json.loads(body.decode() if isinstance(body, bytes) else body)
    return result.get("response", result.get("result", str(result)))

Passing "DEFAULT" as the qualifier calls the live version. To pin to a specific version (staging pin, canary, or rollback), pass a numeric version string instead — see agents-deploy/references/versioning.md for the full workflow.

For Strands, register it as a @tool:

from strands import Agent, tool

@tool
def delegate_to_specialist(task: str) -> str:
    """
    Delegate a complex analysis task to the specialist agent.
    Use this when the task requires deep domain expertise.

    Args:
        task: The specific task or question for the specialist.

    Returns:
        The specialist's detailed response.
    """
    return call_specialist(task)

@app.entrypoint
def invoke(payload, context):
    agent = Agent(
        model=load_model(),  # scaffolded by `agentcore create`
        system_prompt="""You are an orchestrator. For complex analysis tasks,
        delegate to the specialist using the delegate_to_specialist tool.
        Synthesize the specialist's response for the user.""",
        tools=[delegate_to_specialist],
    )
    result = agent(payload.get("prompt", ""))
    return {"response": str(result)}

if __name__ == "__main__":
    app.run()

For LangGraph, add it as a tool node:

from langchain_core.tools import tool as lc_tool

@lc_tool
def delegate_to_specialist(task: str) -> str:
    """Delegate complex tasks to the specialist agent."""
    return call_specialist(task)

# Add to your LangGraph tool node
tools = [delegate_to_specialist]
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)

For OpenAI Agents SDK, register as a @function_tool:

from agents import Agent, Runner, function_tool

@function_tool
def delegate_to_specialist(task: str) -> str:
    """Delegate a complex analysis task to the specialist agent.
    Use when the task requires deep domain expertise."""
    return call_specialist(task)

@app.entrypoint
async def invoke(payload, context):
    agent = Agent(
        name="Orchestrator",
        instructions="For complex analysis, delegate to the specialist using delegate_to_specialist. Synthesize the response for the user.",
        tools=[delegate_to_specialist],
    )
    result = await Runner.run(agent, payload["prompt"])
    return {"response": result.final_output}

For Google ADK, pass as a plain function in the agent's tools=[] list. Note: the official samples use A2A for ADK multi-agent patterns (see awslabs/agentcore-samples/02-use-cases/A2A-multi-agent-incident-response/host_adk_agent/). The direct-invocation pattern below is extrapolated from the ADK base template — validate against your ADK version before relying on it in production:

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

def delegate_to_specialist(task: str) -> str:
    """Delegate complex analysis to the specialist agent."""
    return call_specialist(task)

agent = Agent(
    model="gemini-2.5-flash",
    name="orchestrator",
    description="Orchestrator that delegates complex tasks to specialists.",
    instruction="For complex analysis, call delegate_to_specialist and synthesize the response.",
    tools=[delegate_to_specialist],
)

@app.entrypoint
async def invoke(payload, context):
    user_id = payload.get("user_id", "default_user")
    session_id = getattr(context, "session_id", "default_session")
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="orchestrator", user_id=user_id, session_id=session_id
    )
    runner = Runner(agent=agent, app_name="orchestrator", session_service=session_service)
    content = types.Content(role="user", parts=[types.Part(text=payload["prompt"])])
    async for event in runner.run_async(user_id=user_id, session_id=session.id, new_message=content):
        if event.is_final_response():
            return {"response": event.content.parts[0].text}

For a validated ADK multi-agent pattern, use A2A instead of direct invocation — see the A2A section below and the sample linked above.

For Claude Agent SDK: See awslabs/agentcore-samples/03-integrations/agentic-frameworks/claude-agent/claude-sub-agents/ for the official sub-agent pattern. This plugin doesn't ship a Claude SDK delegation pattern because the sample is more current than anything we could extrapolate.

Step 3: Grant IAM permission

The orchestrator's execution role needs permission to invoke the specialist:

{
  "Effect": "Allow",
  "Action": "bedrock-agentcore:InvokeAgentRuntime",
  "Resource": "arn:aws:bedrock-agentcore:<REGION>:<YOUR_ACCOUNT_ID>:runtime/SpecialistAgent-*"
}

Add this to agentcore/agentcore.json under the orchestrator agent's IAM config, or add it manually to the auto-created execution role after deploy.

Step 4: Pass the specialist ARN at deploy time

Add the specialist ARN as an environment variable in the orchestrator's deployment:

# After deploying the specialist, get its ARN:
SPECIALIST_ARN=$(agentcore status --runtime SpecialistAgent --json | jq -r '.runtimes[0].arn')

# For local dev, write to .env.local:
echo "SPECIALIST_AGENT_ARN=$SPECIALIST_ARN" >> agentcore/.env.local

For the deployed orchestrator, the specialist ARN needs to be available as an environment variable. The recommended pattern is:

1. Edit agentcore/agentcore.json — find the orchestrator agent's entry and add the env var to its configuration (the exact field name depends on your CLI version; run agentcore validate after editing). In current CLI versions, agent environment variables are typically managed through the deployment config.

2. Or use CDK overrides — for teams using the CDK constructs directly, set the env var in the Runtime construct's environment property.

3. Or write the env var at deploy time — some teams use a pre-deploy script that generates agentcore/.env.local and agentcore/agentcore.json updates together:

# pre-deploy.sh — run before every orchestrator deploy
SPECIALIST_ARN=$(agentcore status --runtime SpecialistAgent --json | jq -r '.runtimes[0].arn')
echo "SPECIALIST_AGENT_ARN=$SPECIALIST_ARN" >> agentcore/.env.local

# Then deploy
agentcore deploy -y

The CLI does not currently provide a dedicated --env flag on agentcore add agent. Check agentcore add agent --help for the current options in your CLI version.

Pattern 2: A2A protocol

The specialist exposes the A2A standard — discoverable via an agent card, callable via JSON-RPC. AgentCore's A2A runtime handles the HTTP server, port binding (9000), and agent card serving for you.

Step 1: Build the A2A specialist

Use the serve_a2a helper from bedrock-agentcore — this matches what the CLI scaffolds via agentcore create --protocol A2A.

# app/SpecialistA2A/main.py
from strands import Agent, tool
from strands.multiagent.a2a.executor import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a
from model.load import load_model


@tool
def analyze_data(dataset_name: str) -> str:
    """Run detailed analysis on the named dataset."""
    # Your specialist logic here
    return f"Analysis results for {dataset_name}..."


agent = Agent(
    model=load_model(),
    system_prompt="You are an analysis specialist. Use tools when appropriate.",
    tools=[analyze_data],
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))

# requirements.txt
strands-agents[a2a]
bedrock-agentcore

serve_a2a handles port 9000 binding, agent card generation at /.well-known/agent-card.json, and JSON-RPC routing automatically. No FastAPI or uvicorn needed.

Step 2: Deploy the A2A specialist

agentcore create --name SpecialistA2A --protocol A2A
# The CLI scaffolds app/SpecialistA2A/main.py with the serve_a2a pattern shown above — customize it with your specialist logic
agentcore deploy -y

After deploy, get the runtime URL:

agentcore fetch access --name SpecialistA2A --type agent

Step 3: Test locally

# Start the A2A server locally (from your project dir)
agentcore dev

# Test the agent card (discovery)
curl http://localhost:9000/.well-known/agent-card.json | jq .

# Send a message
curl -X POST http://localhost:9000 \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "req-001",
    "method": "message/send",
    "params": {
      "message": {
        "role": "user",
        "parts": [{"kind": "text", "text": "What is 42 * 17?"}],
        "messageId": "msg-001"
      }
    }
  }' | jq .

Step 4: Call the A2A specialist from the orchestrator

The specialist URL is a non-secret identifier, so pass it via an env var in the orchestrator's deployment config. The bearer token is a secret — do not stash it in os.getenv(...) on the deployed runtime (runtime env vars are not vault-backed). Register an OAuth M2M provider once, then use @requires_access_token to fetch a fresh token at call time:

# One-time: register the OAuth provider that issues tokens for the specialist.
# Omit --client-secret to get an interactive prompt (value goes straight into the credential provider).
agentcore add credential \
  --name SpecialistA2A \
  --type oauth \
  --discovery-url https://<YOUR_IDP>/.well-known/openid-configuration \
  --client-id <CLIENT_ID> \
  --scopes a2a.invoke

import asyncio
import os
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TextPart
from bedrock_agentcore.identity.auth import requires_access_token

# Non-secret identifier — fine to pull from the environment.
SPECIALIST_URL = os.getenv("SPECIALIST_A2A_URL")

@requires_access_token(
    provider_name="SpecialistA2A",
    scopes=["a2a.invoke"],
    auth_flow="M2M",
)
async def call_a2a_specialist(message: str, *, access_token: str) -> str:
    session_id = str(uuid4())
    headers = {
        "Authorization": f"Bearer {access_token}",
        "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": session_id,
    }

    async with httpx.AsyncClient(timeout=300, headers=headers) as http_client:
        resolver = A2ACardResolver(httpx_client=http_client, base_url=SPECIALIST_URL)
        agent_card = await resolver.get_agent_card()

        config = ClientConfig(httpx_client=http_client, streaming=False)
        client = ClientFactory(config).create(agent_card)

        msg = Message(
            kind="message",
            role=Role.user,
            parts=[Part(TextPart(kind="text", text=message))],
            message_id=uuid4().hex,
        )

        async for event in client.send_message(msg):
            if hasattr(event, "parts"):
                return " ".join(p.text for p in event.parts if hasattr(p, "text"))
    return ""

# Use in your orchestrator's entrypoint:
@app.entrypoint
def invoke(payload, context):
    result = asyncio.run(call_a2a_specialist(payload.get("prompt", "")))
    return {"response": result}

The decorator handles caching and refresh. For local dev, put the OAuth values in agentcore/.env.local so agentcore dev can resolve the decorator — the deployed runtime reads them from the credential provider instead.

Shared memory across agents

Memory is a top-level resource — not nested under a single agent. Multiple agents can share it by reading the same env var.

Setup

1. Create one shared memory resource:

agentcore add memory --name SharedMemory --strategies SEMANTIC,USER_PREFERENCE

1. In each agent's code, read the same env var:

MEMORY_ID = os.getenv("MEMORY_SHAREDMEMORY_ID")

1. Use a consistent actor_id scheme — typically the end user's ID — so both agents read and write the same user's memory.

Key consideration

When multiple agents share memory, they share the same namespace. Use namespaced paths to avoid collisions:

# Orchestrator writes to /orchestrator/ namespace
memory_client.create_event(
    memory_id=MEMORY_ID,
    actor_id=user_id,
    session_id=session_id,
    messages=[("User asked about X", "user")],
)

# Specialist reads from all namespaces
turns = memory_client.get_last_k_turns(
    memory_id=MEMORY_ID,
    actor_id=user_id,
    session_id=session_id,
    k=5,
)

Troubleshooting

A2A server not responding:

• Verify it's running on port 9000 (not 8080)
• Check the agent card endpoint returns: curl http://localhost:9000/.well-known/agent-card.json
• Verify your main.py uses serve_a2a(StrandsA2AExecutor(agent)) — the older A2AServer + FastAPI pattern is deprecated in favor of this

Direct invocation permission denied:

• Check the orchestrator's execution role has bedrock-agentcore:InvokeAgentRuntime
• Verify the resource ARN pattern matches the specialist's ARN
• IAM changes take ~30 seconds to propagate

Specialist not found:

• Verify SPECIALIST_AGENT_ARN env var is set correctly
• Check agentcore status --runtime SpecialistAgent shows deployed state

A2A auth errors:

• A2A supports SigV4 and OAuth 2.0 — make sure you're using the right auth method
• Get the correct bearer token: agentcore fetch access --name SpecialistA2A --type agent

Output

• Decision tree to choose the right pattern
• Complete code for the chosen pattern (orchestrator + specialist)
• IAM policy for agent-to-agent invocation
• Local testing commands

Quality criteria

• Pattern recommendation matches the developer's latency and interoperability needs
• Generated code includes correct IAM permissions for agent-to-agent invocation
• A2A server runs on port 9000 (not 8080) using serve_a2a(StrandsA2AExecutor(agent))
• Agent card is at /.well-known/agent-card.json with correct capabilities
• Shared memory uses consistent actor_id scheme across agents