Agents Build

/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

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

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

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)))

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()

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)

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}

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}

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

# 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

# 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

# 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

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

agentcore fetch access --name SpecialistA2A --type agent

# 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 .

# 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}

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

MEMORY_ID = os.getenv("MEMORY_SHAREDMEMORY_ID")

# 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,
)