How agents
talk

[
  {
    "uri": "file:///Users/planner/conference-2025",
    "name": "2025 Conference Workspace"
  }
]

"""
MCP Server — exposes all the MCP primitives over stdio transport:

  * Tools      — get_weather, list_client_roots
  * Resources  — greeting://{name}  and  courses://directory (JSON)
  * Prompts    — weather_report, generate_study_plan

MCP (Model Context Protocol) lets LLM applications discover and call tools,
read resources, and use prompt templates through one standard protocol.

Run:  python client.py   (the client starts this server for you)
"""

import json

# FastMCP: high-level helper for building MCP servers. It handles JSON-RPC
# parsing, registration of tools/resources/prompts, and the transport.
from mcp.server.fastmcp import Context, FastMCP

# The name identifies this server to clients when they connect.
mcp = FastMCP("WeatherServer", log_level="WARNING")

# --- Tool: get_weather ---------------------------------------------------
# Tools are functions an LLM can call. @mcp.tool() registers the function so
# it appears in list_tools() and can be invoked with call_tool(). The
# docstring becomes the description the LLM sees.

@mcp.tool()
def get_weather(city: str) -> str:
    """Get the current weather for a given city."""
    # Fake data for demo purposes — a real server would call a weather API
    fake_weather = {
        "london": "15°C, cloudy",
        "paris": "18°C, sunny",
        "tokyo": "22°C, humid",
        "new york": "12°C, windy",
        "bangalore": "26°C, pleasant with light showers",
    }
    result = fake_weather.get(city.lower())
    if result:
        return f"Weather in {city}: {result}"
    return f"Sorry, no weather data available for '{city}'."

# --- Tool that uses a client capability: roots ---------------------------
# Roots are folders/URIs the *client* says the server may work within.
# A server can ask the client for its roots at any time.

@mcp.tool()
async def list_client_roots(ctx: Context) -> str:
    """List the workspace roots the connected client has shared with this server."""
    result = await ctx.session.list_roots()
    if not result.roots:
        return "The client did not share any roots."
    return "\n".join(f"{root.name}: {root.uri}" for root in result.roots)

# --- Resource: greeting ---------------------------------------------------
# Resources are data an LLM application can read (like GET endpoints).
# "greeting://{name}" is a URI template: read_resource("greeting://Developer")
# fills in {name}. Use resources for data, tools for actions.

@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Return a personalized greeting."""
    return f"Hello, {name}! Welcome to the WeatherServer."

# --- Resource: course directory (structured JSON) -------------------------
# A fixed URI with a MIME type tells the client how to interpret the content.

COURSES = [
    {"course": "Machine Learning", "instructor": "Dr. Rao", "weeks": 12},
    {"course": "Agentic AI on AWS", "instructor": "Shivank Agrawal", "weeks": 8},
    {"course": "Data Engineering", "instructor": "Ms. Iyer", "weeks": 10},
]


@mcp.resource(
    "courses://directory",
    name="courseDirectory",
    title="Course Directory",
    description="List of courses and instructors in a program",
    mime_type="application/json",
)
def course_directory() -> str:
    return json.dumps(COURSES, indent=2)

# --- Prompts ---------------------------------------------------------------
# Prompts are reusable templates. Clients call get_prompt(name, arguments)
# and get back ready-to-send messages.

@mcp.prompt()
def weather_report(city: str) -> str:
    """Generate a prompt asking for a weather report."""
    return f"Please provide a detailed weather report for {city}, including temperature, humidity, and forecast."


@mcp.prompt(title="Generate Study Plan")
def generate_study_plan(course: str, exam_date: str) -> str:
    """Create a personalized study plan for a student based on their course and exam date."""
    return (
        f"Create a week-by-week study plan for the course '{course}'. "
        f"The exam is on {exam_date}. Include revision weeks, practice tests, "
        "and one concrete goal per week."
    )

# --- Start the server ------------------------------------------------------
# "stdio" = read JSON-RPC messages from stdin, write responses to stdout.
# The client spawns this script as a subprocess and talks to it through pipes.

if __name__ == "__main__":
    mcp.run(transport="stdio")

"""
MCP Client — starts server.py over stdio and exercises every MCP primitive:
tools, resources, prompts, and roots.

Run:  python client.py
"""

import asyncio
import json
import sys
from pathlib import Path

import mcp.types as types

# ClientSession: the MCP protocol session (initialize, list, call, ...)
# StdioServerParameters: how to start the server subprocess
from mcp import ClientSession, StdioServerParameters

# stdio_client: starts the server process and returns read/write streams
from mcp.client.stdio import stdio_client

HERE = Path(__file__).resolve().parent

# --- Roots -----------------------------------------------------------------
# The client decides which folders the server may work within. When the
# server asks (list_roots), this callback answers.

async def list_roots(context) -> types.ListRootsResult:
    return types.ListRootsResult(
        roots=[
            types.Root(uri=HERE.as_uri(), name="Course workspace"),
        ]
    )

async def main():
    # Start "python server.py" as a subprocess. sys.executable is the same
    # Python that runs this client, so the server sees the same packages.
    server_params = StdioServerParameters(
        command=sys.executable,
        args=[str(HERE / "server.py")],
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream, list_roots_callback=list_roots) as session:
            # Required first step: exchange protocol version and capabilities
            await session.initialize()

            # --- 1. Discover tools ----------------------------------------
            tools = await session.list_tools()
            print("=== Available Tools ===")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")

            # --- 2. Discover resources ------------------------------------
            # Fixed resources and URI templates are listed separately.
            print("\n=== Available Resources ===")
            resources = await session.list_resources()
            for r in resources.resources:
                print(f"  - {r.uri} ({r.mimeType}): {r.description}")
            templates = await session.list_resource_templates()
            for rt in templates.resourceTemplates:
                print(f"  - {rt.uriTemplate}: {rt.description}")

            # --- 3. Discover prompts --------------------------------------
            print("\n=== Available Prompts ===")
            prompts = await session.list_prompts()
            for p in prompts.prompts:
                args = ", ".join(a.name for a in (p.arguments or []))
                print(f"  - {p.name}({args}): {p.description}")

            # --- 4. Call a tool (known city) ------------------------------
            print("\n=== Calling get_weather('London') ===")
            result = await session.call_tool("get_weather", {"city": "London"})
            print(f"  Result: {result.content[0].text}")

            # --- 5. Call a tool (unknown city) ----------------------------
            print("\n=== Calling get_weather('Berlin') ===")
            result = await session.call_tool("get_weather", {"city": "Berlin"})
            print(f"  Result: {result.content[0].text}")

            # --- 6. Read a templated resource -----------------------------
            print("\n=== Reading greeting://Developer ===")
            resource = await session.read_resource("greeting://Developer")
            print(f"  Result: {resource.contents[0].text}")

            # --- 7. Read a JSON resource ----------------------------------
            print("\n=== Reading courses://directory ===")
            resource = await session.read_resource("courses://directory")
            courses = json.loads(resource.contents[0].text)
            for c in courses:
                print(f"  - {c['course']} — {c['instructor']} ({c['weeks']} weeks)")

            # --- 8. Get prompts -------------------------------------------
            print("\n=== Getting weather_report prompt ===")
            prompt = await session.get_prompt("weather_report", {"city": "Tokyo"})
            print(f"  Result: {prompt.messages[0].content.text}")

            print("\n=== Getting generate_study_plan prompt ===")
            prompt = await session.get_prompt(
                "generate_study_plan", {"course": "Machine Learning", "exam_date": "2026-05-10"}
            )
            print(f"  Result: {prompt.messages[0].content.text}")

            # --- 9. Roots: the server asks the client ---------------------
            print("\n=== Calling list_client_roots ===")
            result = await session.call_tool("list_client_roots", {})
            print(f"  Result: {result.content[0].text}")


if __name__ == "__main__":
    asyncio.run(main())

pip install -r requirements.txt
python client.py        # starts server.py automatically

mcp_client = MCPClient(lambda: stdio_client(StdioServerParameters(command=..., args=[...])))

with mcp_client:
    tools = mcp_client.list_tools_sync()
    agent = Agent(tools=tools)
    agent("...")

"""
A Strands agent that gets its tools from an MCP server.

It starts the server from ../mcp-basics/server.py over stdio, discovers its
tools, and lets the model decide when to call them. The agent code knows
nothing about the weather tool in advance — it all comes over MCP.

Run:  python agent_with_local_mcp.py
"""

import os
import sys
from pathlib import Path

from mcp import StdioServerParameters, stdio_client
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.tools.mcp import MCPClient

MODEL_ID = os.getenv("BEDROCK_MODEL_ID", "global.anthropic.claude-sonnet-4-6")
REGION = os.getenv("AWS_REGION", os.getenv("AWS_DEFAULT_REGION", "us-east-1"))

SERVER = Path(__file__).resolve().parent.parent / "mcp-basics" / "server.py"

# MCPClient takes a function that opens the transport. Here: spawn the server.
mcp_client = MCPClient(
    lambda: stdio_client(StdioServerParameters(command=sys.executable, args=[str(SERVER)]))
)

with mcp_client:
    tools = mcp_client.list_tools_sync()
    print("Tools discovered over MCP:", [t.tool_name for t in tools])

    agent = Agent(model=BedrockModel(model_id=MODEL_ID, region_name=REGION), tools=tools)
    agent("I'm flying from Bangalore to London tomorrow. What's the weather like in both cities?")
    print()

"""
A Strands agent connected to the AWS Documentation MCP server.

The AWS Labs documentation server (https://github.com/awslabs/mcp) lets the
agent search and read official AWS docs, so its answers come from current
documentation instead of only the model's training data.

Prerequisite: `uv` installed (it provides the `uvx` command that downloads and
runs the server on demand).

Run:  python agent_with_aws_docs_mcp.py
"""

import os
import shutil

from mcp import StdioServerParameters, stdio_client
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.tools.mcp import MCPClient

MODEL_ID = os.getenv("BEDROCK_MODEL_ID", "global.anthropic.claude-sonnet-4-6")
REGION = os.getenv("AWS_REGION", os.getenv("AWS_DEFAULT_REGION", "us-east-1"))

if shutil.which("uvx") is None:
    raise SystemExit("`uvx` not found. Install uv first: pip install uv")

# Create MCP client
mcp_client = MCPClient(
    lambda: stdio_client(
        StdioServerParameters(
            command="uvx",
            args=["awslabs.aws-documentation-mcp-server@latest"],
            env={**os.environ, "FASTMCP_LOG_LEVEL": "ERROR"},
        )
    ),
    startup_timeout=120,  # first run downloads the server package
)

# Use with agent
with mcp_client:
    tools = mcp_client.list_tools_sync()
    print("Tools from the AWS Documentation MCP server:", [t.tool_name for t in tools])

    agent = Agent(
        model=BedrockModel(model_id=MODEL_ID, region_name=REGION),
        tools=tools,
        system_prompt="Answer AWS questions using the documentation tools. Cite the doc URLs you used.",
    )
    agent("What is AWS Lambda, and what is the maximum timeout for a Lambda function?")
    print()

{
  "name": "Weather Agent",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "get_weather",
      "name": "Get Weather",
      "description": "Returns current weather conditions for a given city."
    }
  ],
  "url": "https://api.example.com/weather-agent/"
}

"""
Weather Agent — A2A Server on port 9001.

Exposes a weather lookup tool via the Strands A2A protocol.

Run:  python weather_agent_server.py
"""

import logging
import os

# Agent: the core Strands agent class that wraps an LLM with tools
# tool: decorator that registers a Python function as a tool the LLM can call
from strands import Agent, tool

# A2AServer: wraps a Strands Agent into an HTTP server that speaks the A2A protocol
from strands.multiagent.a2a import A2AServer

# AgentSkill: A2A SDK type that describes a capability in the agent card
from a2a.types import AgentSkill

# BedrockModel: the LLM behind this agent (Amazon Bedrock)
from strands.models.bedrock import BedrockModel

MODEL_ID = os.getenv("BEDROCK_MODEL_ID", "global.anthropic.claude-sonnet-4-6")
REGION = os.getenv("AWS_REGION", os.getenv("AWS_DEFAULT_REGION", "us-east-1"))

# Set up logging so we can see what's happening during requests
logging.basicConfig(level=logging.INFO)

# --- Fake Weather Data ---------------------------------------------------
# In a real app this would call a weather API.
# Here we use a dictionary for demo purposes.

FAKE_WEATHER = {
    "london": "15°C, cloudy with occasional rain",
    "paris": "18°C, sunny with light breeze",
    "tokyo": "22°C, humid and partly cloudy",
    "new york": "12°C, windy and cool",
    "dubai": "38°C, hot and sunny",
    "sydney": "20°C, mild and clear",
    "cancun": "30°C, tropical and sunny",
    "rome": "24°C, warm and sunny",
    "bangkok": "33°C, hot and humid",
    "reykjavik": "5°C, cold and overcast",
}


# --- Weather Tool --------------------------------------------------------
# The @tool decorator registers this function so the LLM can call it.
# The docstring becomes the tool description the LLM sees when deciding
# which tool to use. Args/Returns are parsed for the tool's input schema.

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a given city.

    Args:
        city: Name of the city to get weather for.

    Returns:
        A string describing the current weather conditions.
    """
    # Normalize input: strip whitespace, lowercase for dictionary lookup
    result = FAKE_WEATHER.get(city.strip().lower())
    if result:
        return f"Weather in {city}: {result}"
    # If city not found, tell the user what cities are available
    return f"No weather data available for '{city}'. Available cities: {', '.join(FAKE_WEATHER.keys())}"

# --- Agent Card Skills ---------------------------------------------------
# Skills are published in the agent card at /.well-known/agent-card.json.
# Any A2A client fetches this card first to discover what the agent can do
# BEFORE sending any messages. Think of it like an API spec for agents.

WEATHER_SKILLS = [
    AgentSkill(
        # Unique identifier for this skill
        id="get_weather",
        # Human-readable name shown in the agent card
        name="Get Weather",
        # Detailed description — helps clients understand what this skill does
        description=(
            "Returns current weather conditions (temperature, sky, wind) "
            "for a given city. Supports: London, Paris, Tokyo, New York, "
            "Dubai, Sydney, Cancun, Rome, Bangkok, Reykjavik."
        ),
        # Tags for categorization and discovery
        tags=["weather", "temperature", "forecast", "travel"],
        # Example prompts — hints to clients on how to use this skill
        examples=[
            "What is the weather in London?",
            "How hot is it in Dubai right now?",
            "Tell me the weather conditions in Tokyo",
        ],
    ),
]

# --- Strands Agent -------------------------------------------------------
# This is the actual AI agent. It combines:
#   - An LLM (Amazon Bedrock, chosen by BEDROCK_MODEL_ID)
#   - A system prompt (instructions for the LLM)
#   - Tools (functions the LLM can call)
#
# When a message comes in, the agent sends it to the LLM. The LLM reads
# the system prompt, sees the available tools, and decides whether to call
# get_weather. If it does, the tool runs, and the LLM formats the final answer.

weather_agent = Agent(
    # The LLM that powers this agent
    model=BedrockModel(model_id=MODEL_ID, region_name=REGION),
    # Name and description — used in the A2A agent card
    name="Weather Agent",
    description="Provides current weather information for cities around the world.",
    # System prompt — instructions the LLM follows for every request
    system_prompt=(
        "You are a helpful weather assistant. Use the get_weather tool to look up "
        "weather conditions for any city the user asks about. Be concise and friendly."
    ),
    # Register the get_weather tool so the LLM can call it
    tools=[get_weather],
    # Disable streaming callback on the server side (responses are collected, not printed)
    callback_handler=None,
)

# --- A2A Server ----------------------------------------------------------
# Only runs when you execute this file directly (python weather_agent_server.py).
# Importing this file from another module won't start the server.

if __name__ == "__main__":
    # A2AServer wraps the Strands Agent into an HTTP server that:
    #   1. Serves an agent card at GET /.well-known/agent-card.json
    #   2. Accepts A2A messages at POST / (JSON-RPC: message/send, message/sendStream)
    #   3. Routes messages through the Strands Agent (LLM + tools)
    #   4. Returns responses as A2A task objects with artifacts
    server = A2AServer(
        # The Strands Agent that handles all incoming messages
        agent=weather_agent,
        # Bind to localhost only (use "0.0.0.0" to expose to network)
        host="127.0.0.1",
        # HTTP port to listen on
        port=9001,
        # Skills published in the agent card for client discovery
        skills=WEATHER_SKILLS,
        # Version string in the agent card
        version="1.0.0",
        # Use A2A-spec-compliant SSE streaming with artifact updates
        # (vs legacy status-only streaming). The agent card will advertise
        # {"capabilities": {"streaming": true}} so clients know SSE is supported.
        enable_a2a_compliant_streaming=True,
    )
    print("Weather Agent A2A server starting on http://127.0.0.1:9001")
    # Start the uvicorn HTTP server. This is a blocking call —
    # the process stays alive serving requests until you Ctrl+C.
    server.serve()

{
  "name": "Weather Agent",
  "description": "Provides current weather information for cities around the world.",
  "version": "1.0.0",
  "protocolVersion": "0.3.0",
  "preferredTransport": "JSONRPC",
  "capabilities": { "streaming": true },
  "skills": [
    {
      "id": "get_weather",
      "name": "Get Weather",
      "description": "Returns current weather conditions...",
      "tags": ["weather", "temperature", "forecast", "travel"],
      "examples": ["What is the weather in London?", "..."]
    }
  ],
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text"],
  "url": "http://127.0.0.1:9001/"
}

# --- Fake Flight Data ----------------------------------------------------
# In a real app this would query a flights API or database.
# Here we use a dictionary keyed by (origin, destination) tuples.

FAKE_FLIGHTS = {
    ("london", "paris"): [
        {"airline": "AirEurope", "flight": "AE101", "depart": "08:00", "arrive": "10:30", "price": "$120"},
        {"airline": "SkyWings", "flight": "SW202", "depart": "14:00", "arrive": "16:30", "price": "$95"},
    ],
    ("new york", "london"): [
        {"airline": "TransAtlantic", "flight": "TA501", "depart": "22:00", "arrive": "10:00+1", "price": "$450"},
        {"airline": "AirEurope", "flight": "AE303", "depart": "18:00", "arrive": "06:00+1", "price": "$380"},
    ],
    ("tokyo", "sydney"): [
        {"airline": "PacificAir", "flight": "PA701", "depart": "09:00", "arrive": "20:00", "price": "$520"},
    ],
    ("paris", "rome"): [
        {"airline": "AirEurope", "flight": "AE150", "depart": "07:30", "arrive": "09:30", "price": "$85"},
        {"airline": "MedFly", "flight": "MF44", "depart": "12:00", "arrive": "14:00", "price": "$70"},
    ],
    ("dubai", "bangkok"): [
        {"airline": "GulfAir", "flight": "GA880", "depart": "02:00", "arrive": "12:00", "price": "$340"},
    ],
    ("new york", "cancun"): [
        {"airline": "SunJet", "flight": "SJ610", "depart": "06:00", "arrive": "10:00", "price": "$250"},
        {"airline": "SkyWings", "flight": "SW415", "depart": "11:00", "arrive": "15:00", "price": "$210"},
    ],
}

# --- Flight Tools --------------------------------------------------------
# Each @tool-decorated function becomes a tool the LLM can call.
# The docstring is what the LLM sees when deciding which tool to use.
# Args/Returns are parsed to build the tool's input schema automatically.


@tool
def search_flights(origin: str, destination: str) -> str:
    """Search for available flights between two cities.

    Args:
        origin: Departure city name.
        destination: Arrival city name.

    Returns:
        A formatted list of available flights or a message if none found.
    """
    # Normalize both cities to lowercase for dictionary lookup
    key = (origin.strip().lower(), destination.strip().lower())
    flights = FAKE_FLIGHTS.get(key)

    # If no flights found, show the user what routes are available
    if not flights:
        available = [f"{o.title()} -> {d.title()}" for o, d in FAKE_FLIGHTS.keys()]
        return (
            f"No flights found from {origin} to {destination}. "
            f"Available routes: {', '.join(available)}"
        )

    # Format each flight as a readable line
    lines = [f"Flights from {origin.title()} to {destination.title()}:"]
    for f in flights:
        lines.append(
            f"  {f['airline']} {f['flight']}: depart {f['depart']}, "
            f"arrive {f['arrive']}, price {f['price']}"
        )
    return "\n".join(lines)


@tool
def book_flight(flight_number: str, passenger_name: str) -> str:
    """Book a specific flight for a passenger.

    Args:
        flight_number: The flight number to book (e.g. AE101).
        passenger_name: Name of the passenger.

    Returns:
        Booking confirmation or error message.
    """
    # Search all routes for the matching flight number
    for flights in FAKE_FLIGHTS.values():
        for f in flights:
            if f["flight"].upper() == flight_number.strip().upper():
                # Generate a fake confirmation number from a hash
                return (
                    f"Booking confirmed! {passenger_name} is booked on "
                    f"{f['airline']} flight {f['flight']} "
                    f"(depart {f['depart']}, arrive {f['arrive']}). "
                    f"Total: {f['price']}. Confirmation #: TRV-{hash(flight_number + passenger_name) % 100000:05d}"
                )
    return f"Flight {flight_number} not found. Please search for available flights first."

"""
Travel Orchestrator — A2A client that talks to Weather + Flights agents.

This is the A2A CLIENT side. It connects to two remote A2A agent servers,
sends them messages using the A2A protocol, and prints the results.

No LLM here — just direct A2A request/response.

Prerequisites:
    1. Start weather agent:  python weather_agent_server.py
    2. Start flights agent:  python flights_agent_server.py
    3. Run this:             python travel_orchestrator.py

Usage:
    python travel_orchestrator.py
    python travel_orchestrator.py "London" "New York" "London"
"""

import sys
import asyncio
import logging
from uuid import uuid4

# httpx: async HTTP client used to make requests to the A2A servers
import httpx

# A2ACardResolver: fetches the agent card from /.well-known/agent-card.json
# ClientConfig: configures the A2A client (streaming vs sync, timeout, etc.)
# ClientFactory: creates an A2A client from an agent card
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory

# A2A message types used to construct the request
# Message: the envelope for a user or agent message
# Part/TextPart: wraps the actual text content
# Role: identifies the sender (user or agent)
from a2a.types import Message, Part, Role, TextPart

# Set up logging so we can see A2A protocol activity
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# URLs of the two A2A agent servers
WEATHER_URL = "http://127.0.0.1:9001"
FLIGHTS_URL = "http://127.0.0.1:9002"

# Timeout for HTTP requests (5 minutes — agents may take time with LLM calls)
TIMEOUT = 300

# --- Helper: Build an A2A Message ----------------------------------------
# Creates a standard A2A Message object that the server expects.
# This is the equivalent of a user typing a question in a chat.

def make_message(text: str) -> Message:
    return Message(
        kind="message",          # Always "message" for chat messages
        role=Role.user,          # We're the user sending a question
        parts=[Part(TextPart(    # Wrap the text in a TextPart inside a Part
            kind="text",
            text=text,
        ))],
        message_id=uuid4().hex,  # Unique ID for this message
    )


# --- Helper: Extract Text from A2A Response ------------------------------
# A2A servers return a "task" object. The actual answer text lives inside
# task.artifacts[].parts[].text. This function digs it out.

def extract_text(result) -> str:
    """Pull text out of an A2A task response (from artifacts)."""
    if hasattr(result, "artifacts") and result.artifacts:
        texts = []
        for artifact in result.artifacts:
            for part in artifact.parts:
                # The a2a SDK wraps parts in a discriminated union (Part.root)
                if hasattr(part, "root") and hasattr(part.root, "text"):
                    texts.append(part.root.text)
                elif hasattr(part, "text"):
                    texts.append(part.text)
        if texts:
            return "".join(texts)
    # Fallback: return the raw string representation
    return str(result)

# --- Core: Send a Message to an A2A Agent --------------------------------
# This is the main A2A client logic. It:
#   1. Fetches the agent card to discover capabilities
#   2. Auto-selects streaming or sync mode based on the card
#   3. Sends the message using the A2A protocol
#   4. Extracts and returns the text response

async def send_a2a_message(base_url: str, text: str) -> str:
    """Send a message to an A2A agent, auto-selecting streaming or sync based on the agent card."""
    async with httpx.AsyncClient(timeout=TIMEOUT) as hc:

        # Step 1: Fetch the agent card from /.well-known/agent-card.json
        # This tells us the agent's name, skills, and whether it supports streaming
        resolver = A2ACardResolver(httpx_client=hc, base_url=base_url)
        card = await resolver.get_agent_card()

        # Step 2: Check the agent card capabilities to decide the mode
        # If the server advertises {"capabilities": {"streaming": true}},
        # we use SSE streaming; otherwise we use sync (single response)
        use_streaming = bool(card.capabilities and card.capabilities.streaming)
        mode = "streaming" if use_streaming else "sync"
        logger.info(f"Connected to: {card.name} — mode: {mode}")

        # Step 3: Create an A2A client configured for this agent
        # ClientConfig sets streaming mode; ClientFactory builds the client from the card
        config = ClientConfig(httpx_client=hc, streaming=use_streaming)
        client = ClientFactory(config).create(card)

        # Step 4: Send the message and process response events
        # In sync mode: yields one task with artifacts (the final answer)
        # In streaming mode: yields intermediate Messages + final task with artifacts
        msg = make_message(text)
        final_text = ""

        async for event in client.send_message(msg):
            # Task with artifacts — this is the final complete response
            if hasattr(event, "artifacts") and event.artifacts:
                final_text = extract_text(event)

            # (Task, UpdateEvent) tuple — check for artifacts on the task
            elif isinstance(event, tuple) and len(event) == 2:
                task, _ = event
                if hasattr(task, "artifacts") and task.artifacts:
                    final_text = extract_text(task)

            # Intermediate streaming Message — we skip these and wait for
            # the final artifacts which contain the complete answer
            elif isinstance(event, Message):
                pass

        return final_text if final_text else "No response received."

# --- Main: Orchestrate Calls to Both Agents ------------------------------
# Sends one request to the Weather Agent and one to the Flights Agent,
# then prints a combined trip summary.

async def main(city: str, origin: str, destination: str):
    print("=" * 60)
    print("  Travel Orchestrator (A2A)")
    print("=" * 60)

    # 1. Ask Weather Agent for current conditions at the destination
    print(f"\n>>> Asking Weather Agent: weather in {city}")
    weather = await send_a2a_message(WEATHER_URL, f"What is the weather in {city}?")
    print(f"<<< Weather Agent response:\n{weather}")

    # 2. Ask Flights Agent for available flights on the route
    print(f"\n>>> Asking Flights Agent: flights from {origin} to {destination}")
    flights = await send_a2a_message(FLIGHTS_URL, f"Search flights from {origin} to {destination}")
    print(f"<<< Flights Agent response:\n{flights}")

    # 3. Print a combined trip summary
    print("\n" + "=" * 60)
    print("  Trip Summary")
    print("=" * 60)
    print(f"\nDestination: {city}")
    print(f"Weather: {weather.strip()}")
    print(f"\nFlights ({origin} → {destination}):")
    print(flights.strip())


# --- Entry Point ---------------------------------------------------------
# Supports two modes:
#   - Default: demo with London / New York → London
#   - Custom:  pass city, origin, destination as arguments

if __name__ == "__main__":
    if len(sys.argv) == 4:
        # python travel_orchestrator.py <city> <origin> <destination>
        asyncio.run(main(sys.argv[1], sys.argv[2], sys.argv[3]))
    else:
        # Default demo
        asyncio.run(main("London", "New York", "London"))

pip install -r requirements.txt

# Terminal 1
python weather_agent_server.py

# Terminal 2
python flights_agent_server.py

# Terminal 3 — default demo (New York → London)
python travel_orchestrator.py

# Or specify city, origin, destination
python travel_orchestrator.py "Rome" "Paris" "Rome"

import sys
from pathlib import Path
from mcp import StdioServerParameters, stdio_client
from strands.tools.mcp import MCPClient

# The MCP server from Lab 5.6, started as a child process over stdio
SERVER = Path(__file__).resolve().parent.parent / "mcp-basics" / "server.py"
mcp_client = MCPClient(
    lambda: stdio_client(StdioServerParameters(command=sys.executable, args=[str(SERVER)]))
)

weather_agent = Agent(
    model=BedrockModel(model_id=MODEL_ID, region_name=REGION),
    name="Weather Agent",
    description="Provides current weather information for cities around the world.",
    system_prompt="You are a helpful weather assistant. Be concise and friendly.",
    tools=[mcp_client],      # tools now arrive over MCP instead of @tool
    callback_handler=None,
)