Implementing Multi-Agent Systems with Orchestrator and Specialist Patterns in AWS AgentCore

You can build multi-agent applications in AWS AgentCore using either direct invocation for low-latency internal delegation or the A2A protocol for decoupled, discoverable specialist services.

AWS AgentCore from the aws/agent-toolkit-for-aws repository provides native support for orchestrator-specialist patterns, allowing you to decompose complex workloads across multiple agents. The toolkit implements two primary delegation mechanisms—direct runtime invocation and the Agent-to-Agent (A2A) protocol—both documented in the multi-agent reference implementation.

Understanding the Orchestrator and Specialist Architecture

A multi-agent system in AgentCore consists of a master orchestrator agent that routes tasks to specialized specialist agents. The orchestrator focuses on high-level decision making and task routing, while specialists handle domain-specific processing. This architecture requires proper IAM configuration, environment variable injection for agent endpoints, and optional shared memory for context persistence.

According to the source code in plugins/aws-agents/skills/agents-build/references/multi-agent.md, both patterns require the orchestrator's execution role to have permission to call bedrock-agentcore:InvokeAgentRuntime on the specialist's ARN. Additionally, specialists must be deployed before the orchestrator can obtain their runtime identifiers or A2A endpoints.

Deployment Patterns for Multi-Agent Systems

AgentCore supports two distinct patterns for orchestrator-to-specialist communication, each suited to different architectural requirements.

Direct Invocation Pattern

The direct invocation pattern uses the bedrock-agentcore:InvokeAgentRuntime API to call a specialist agent's runtime ARN directly. This approach minimizes latency and overhead, making it ideal for tightly-coupled internal services where the specialist does not need to expose a public endpoint.

Specialist agents deployed with this pattern receive their runtime ARN after deployment, which you can retrieve using agentcore status or agentcore fetch access. The orchestrator stores this ARN in environment variables and invokes the specialist using the AWS SDK for Python (Boto3).

A2A Protocol Pattern

The A2A (Agent-to-Agent) protocol pattern decouples agents through a standardized JSON-RPC interface. Specialists run an A2A server using serve_a2a() that publishes an agent card containing metadata, capabilities, and endpoint information. This pattern enables runtime discovery, versioning, and richer metadata exchange.

The A2A approach requires the specialist to expose a public endpoint, while the orchestrator uses the a2a.client library to resolve agent cards and handle authentication flows automatically.

Building the Orchestrator Agent

The orchestrator implementation varies based on your chosen delegation pattern, though both approaches share common scaffolding generated by the agentcore create command.

Direct Invocation Implementation

First, deploy your specialist agent and obtain its runtime ARN. Then implement a helper function to wrap the AWS SDK calls:

import os
import json
import boto3

SPECIALIST_ARN = os.getenv("SPECIALIST_AGENT_ARN")
REGION = os.getenv("AWS_REGION", "us-east-1")
client = boto3.client("bedrock-agentcore", region_name=REGION)

def call_specialist(prompt: str, session_id: str = None) -> str:
    payload = {"prompt": prompt}
    kwargs = {
        "agentRuntimeArn": SPECIALIST_ARN,
        "qualifier": "DEFAULT",
        "payload": json.dumps(payload).encode(),
    }
    if session_id:
        kwargs["runtimeSessionId"] = session_id
    
    resp = client.invoke_agent_runtime(**kwargs)
    body = resp["response"].read()
    result = json.loads(body.decode() if isinstance(body, bytes) else body)
    return result.get("response", result.get("result", str(result)))

Reference this helper in your orchestrator's tool definition to enable delegation:

from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp

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

app = BedrockAgentCoreApp()

@app.entrypoint
def invoke(payload, context):
    orchestrator = Agent(
        model=load_model(),
        system_prompt=(
            "You are an orchestrator. Delegate complex analysis "
            "to the specialist using the delegate_to_specialist tool."
        ),
        tools=[delegate_to_specialist],
    )
    result = orchestrator(payload.get("prompt", ""))
    return {"response": str(result)}

This implementation registers delegate_to_specialist as a tool available to the orchestrator's reasoning loop, allowing it to decide when to hand off tasks based on complexity or domain requirements.

A2A Protocol Implementation

For A2A-based delegation, the specialist must run serve_a2a(StrandsA2AExecutor(agent)) to expose the JSON-RPC endpoint. The orchestrator then uses the A2A client library with automatic OAuth token acquisition:

import asyncio
import os
import json
import httpx
from uuid import uuid4
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from bedrock_agentcore.identity.auth import requires_access_token

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 client:
        resolver = A2ACardResolver(httpx_client=client, base_url=SPECIALIST_URL)
        card = await resolver.get_agent_card()
        cfg = ClientConfig(httpx_client=client, streaming=False)
        a2a_client = ClientFactory(cfg).create(card)
        
        msg = Message(
            kind="message",
            role=Role.user,
            parts=[Part(TextPart(kind="text", text=message))],
            message_id=uuid4().hex,
        )
        
        async for ev in a2a_client.send_message(msg):
            if hasattr(ev, "parts"):
                return " ".join(p.text for p in ev.parts if hasattr(p, "text"))
    return ""

@app.entrypoint
def invoke(payload, context):
    result = asyncio.run(call_a2a_specialist(payload.get("prompt", "")))
    return {"response": result}

The @requires_access_token decorator handles OAuth token refresh automatically, while the A2ACardResolver discovers the specialist's capabilities and endpoint configuration at runtime.

Enabling Shared Memory Across Agents

Both patterns support shared memory for maintaining context across agent boundaries. Agents can read from and write to a common memory resource using the memory client, ensuring conversation history persists through orchestrator-specialist hand-offs.

Configure the memory ID via environment variables:

MEMORY_ID = os.getenv("MEMORY_SHAREDMEMORY_ID")

The orchestrator writes context before delegation:

memory_client.create_event(
    memory_id=MEMORY_ID,
    actor_id=user_id,
    session_id=session_id,
    messages=[("User asked about X", "user")],
)

The specialist retrieves this context upon activation:

turns = memory_client.get_last_k_turns(
    memory_id=MEMORY_ID,
    actor_id=user_id,
    session_id=session_id,
    k=5,
)

This pattern is implemented in plugins/aws-agents/skills/agents-build/references/multi-agent.md and enables seamless context sharing without requiring the specialist to maintain independent session state.

Choosing the Right Pattern for Your Use Case

Select your delegation mechanism based on architectural requirements:

• Direct invocation delivers low-latency performance for internal, tightly-coupled tasks where specialists reside within the same AWS account and security boundary.
• A2A protocol provides decoupling, discovery, and versioning capabilities essential for public-facing services or microservice-style architectures.
• Shared memory supplements either pattern when multiple agents must maintain consistent user context across hand-offs.
• Version pinning via the qualifier parameter in call_specialist enables fine-grained control over which specialist version receives traffic, supporting canary deployments and rollbacks.

Summary

Implementing multi-agent systems with orchestrator and specialist patterns in AWS AgentCore requires selecting between direct invocation and A2A protocol delegation based on your latency and coupling requirements.

• Deploy specialists using agentcore create and agentcore deploy, then retrieve runtime ARNs or A2A endpoints via agentcore status.
• Configure IAM permissions to allow bedrock-agentcore:InvokeAgentRuntime calls from the orchestrator's execution role.
• Implement the call_specialist helper function for direct invocation, or use A2ACardResolver and @requires_access_token for A2A-based communication.
• Enable shared memory by writing to memory_client before delegation and reading from it in the specialist to maintain conversation context.
• Pin specialist versions using the qualifier parameter in invoke_agent_runtime calls for production stability.

Frequently Asked Questions

What is the difference between direct invocation and A2A patterns in AgentCore?

Direct invocation uses the AWS SDK to call bedrock-agentcore:InvokeAgentRuntime on a specialist's ARN, providing low-latency communication ideal for internal services. The A2A protocol uses HTTP JSON-RPC with agent card discovery, enabling loose coupling, public endpoints, and runtime service discovery. Direct invocation requires the SPECIALIST_AGENT_ARN environment variable, while A2A requires SPECIALIST_A2A_URL.

How do I configure IAM permissions for multi-agent delegation?

The orchestrator's execution role must include a policy allowing bedrock-agentcore:InvokeAgentRuntime on the specialist's ARN. As documented in plugins/aws-agents/skills/agents-build/references/multi-agent.md, this permission enables the orchestrator to invoke the specialist's runtime regardless of whether using direct invocation or A2A patterns.

Can multiple specialist agents share conversation context?

Yes. Agents can share context using the shared memory feature. The orchestrator writes conversation history to a memory resource using memory_client.create_event() with a specific memory_id, and specialists retrieve this context using memory_client.get_last_k_turns(). This works with both direct invocation and A2A patterns.

How do I version control specialist agents in production?

Use the qualifier parameter in your invoke_agent_runtime calls to pin to specific versions. Instead of using "DEFAULT", specify a version qualifier string (e.g., "v1.2.3") to ensure the orchestrator always invokes a specific specialist deployment. This approach supports blue-green deployments and gradual rollbacks without orchestrator code changes.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}