Tutorial: Multi-Agent System with A2A and MCP

Let's build a multi-agent system for automated trip planning with Ragbits. In this tutorial, we'll create:

1. A Flight Finder Agent that searches for available flights
2. A City Explorer Agent that gathers information about destinations
3. An Orchestrator Agent that coordinates both agents to create comprehensive trip plans

What you'll learn:

• How to create specialized agents with tools/function calling
• How to use the Model Context Protocol (MCP) for external data
• How to expose agents through Agent-to-Agent (A2A) protocol
• How to build an orchestrator that manages conversation context

Configuring the environment

Install the latest Ragbits via pip install -U ragbits[a2a,mcp] to follow along.

During development, we will use OpenAI's gpt-4.1 model. To authenticate, Ragbits will look into your OPENAI_API_KEY. You can easily swap this with other providers.

Recommended: Set up OpenTelemetry tracing to understand what's happening under the hood.

OpenTelemetry is an LLMOps tool that natively integrates with Ragbits and offer explainability and experiment tracking. In this tutorial, you can use OpenTelemetry to visualize prompts and optimization progress as traces to understand the Ragbits' behavior better. Check the full setup guide here.

Building the Flight Finder Agent

We start by defining the prompt that will lead this agent.

flight_agent.py

from pydantic import BaseModel
from ragbits.core.prompt import Prompt

class FlightPromptInput(BaseModel):
    """Defines the structured input schema for the flight search prompt."""

    input: str


class FlightPrompt(Prompt[FlightPromptInput]):
    """Prompt for a flight search assistant."""

    system_prompt = """
    You are a helpful travel assistant that finds available flights between two cities.
    """

    user_prompt = """
    {{ input }}
    """


print(FlightPrompt(FlightPromptInput(input="I need to fly from New York to Paris. What flights are available?")).chat)

[{'role': 'system', 'content': 'You are a helpful travel assistant that finds available flights between two cities.'},
{'role': 'user', 'content': 'I need to fly from New York to Paris. What flights are available?'}]

Next, we define a tool that will provide flight information. Note: in a real application, you'd connect to the actual flight APIs:

flight_agent.py

import json

def get_flight_info(departure: str, arrival: str) -> str:
    """
    Returns flight information between two locations.

    Args:
        departure: The departure city.
        arrival: The arrival city.

    Returns:
        A JSON string with mock flight details.
    """
    if "new york" in departure.lower() and "paris" in arrival.lower():
        return json.dumps(
            {
                "from": "New York",
                "to": "Paris",
                "flights": [
                    {"airline": "British Airways", "departure": "10:00 AM", "arrival": "10:00 PM"},
                    {"airline": "Delta", "departure": "1:00 PM", "arrival": "1:00 AM"},
                ],
            }
        )

    return json.dumps({"from": departure, "to": arrival, "flights": "No flight data available"})

This agent will call this function as needed, and the results will be injected back to the conversation.

Now let's create the agent and test it:

flight_agent.py

from ragbits.agents import Agent
from ragbits.core.llms import LiteLLM

llm = LiteLLM(
    model_name="gpt-4.1",
    use_structured_output=True,
)
flight_agent = Agent(llm=llm, prompt=FlightPrompt, tools=[get_flight_info])

async def main() -> None:
    result = await flight_agent.run(FlightPromptInput(input="I need to fly from New York to Paris. What flights are available?"))
    print(result.content)

if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Run it:

python flight_agent.py

A typical response looks like this:

Here are some available flights from New York to Paris:

1. **British Airways**
   - **Departure:** 10:00 AM
   - **Arrival:** 10:00 PM

2. **Delta**
   - **Departure:** 1:00 PM
   - **Arrival:** 1:00 AM

Please note that the results may differ among the runs due to undeterministic nature of LLM.

Try it yourself

You can try to connect this agents to a real flight API, such as aviationstack.

Building the City Explorer Agent

Let's create a City Explorer Agent that gather and synthesize city information from the internet. Again we start with the prompt:

city_explorer_agent.py

from pydantic import BaseModel
from ragbits.core.prompt import Prompt


class CityExplorerPromptInput(BaseModel):
    """Defines the structured input schema for the city explorer prompt."""

    input: str


class CityExplorerPrompt(Prompt[CityExplorerPromptInput]):
    """Prompt for a city explorer assistant."""

    system_prompt = """
    You are a helpful travel assistant that gathers and synthesizes city information from the internet.
    To gather information call mcp fetch server with the URL to the city's wikipedia page.
    Then synthesize the information into a concise summary.

    e.g
    https://en.wikipedia.org/wiki/London
    https://en.wikipedia.org/wiki/Paris
    """

    user_prompt = """
    {{ input }}.
    """

Now define the agent, We will not build an MCP server from scratch, but run an already existing one - Web Fetcher. Start by installing it with:

pip install mcp-server-fetch

city_explorer_agent.py

from ragbits.agents import Agent
from ragbits.agents.mcp import MCPServerStdio
from ragbits.core.llms import LiteLLM
from ragbits.core.prompt import Prompt
async def main() -> None:
    """Runs the city explorer agent."""
    async with MCPServerStdio(
        params={
            "command": "python",
            "args": ["-m", "mcp_server_fetch"],
        },
        client_session_timeout_seconds=60,
    ) as server:
        llm = LiteLLM(
            model_name="gpt-4.1",
            use_structured_output=True,
        )
        city_explorer_agent = Agent(llm=llm, prompt=CityExplorerPrompt, mcp_servers=[server])
        result = await city_explorer_agent.run(CityExplorerPromptInput(input="Tell me something interesting about Paris."))
        print(result.content)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Test this agent by running:

python city_explorer_agent.py

Paris is the capital and largest city of France, located in the Île-de-France region. Renowned for its historical landmarks, the city is a significant cultural and economic center in Europe. Key attractions include the Eiffel Tower, Notre-Dame Cathedral, the Louvre Museum, and the Arc de Triomphe. Known for its romantic ambiance, Paris is often referred to as "The City of Light."

The city is characterized by its extensive urban area and is densely populated, with a population of over 2 million within city limits and approximately 13 million in the metropolitan area as of 2021. Paris is divided into 20 districts, known as arrondissements. The current mayor is Anne Hidalgo. The city's motto is "Fluctuat nec mergitur," meaning "Tossed by the waves but never sunk," reflecting its resilience.

Paris is a hub for art, fashion, gastronomy, and culture, drawing millions of visitors each year who seek to experience its heritage and vibrant lifestyle.

Notice that we didn't have to write any tool, just reuse already existing one. This is the magic of MCP protocol.

Exposing Agents through A2A

Now we need to expose our agents through the Agent-to-Agent (A2A) protocol so they can be called remotely. We'll create agent cards and servers for both agents. Let's start with the flight agent. Update the main function with:

flight_agent.py

from ragbits.agents.a2a.server import create_agent_server

async def main() -> None:
    """Runs the flight agent."""
    flight_agent_card = await flight_agent.get_agent_card(
        name="Flight Info Agent", description="Provides available flight information between two cities.", port=8000
    )
    flight_agent_server = create_agent_server(flight_agent, flight_agent_card, FlightPromptInput)
    await flight_agent_server.serve()

and then run.

python flight_agent.py

Now, your server with agent should be up and running

INFO:     Started server process [1473119]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

You can go to your browser and type http://127.0.0.1:8000/.well-known/agent.json to see and agent card:

{
  "additionalInterfaces": null,
  "capabilities": {
    "extensions": null,
    "pushNotifications": null,
    "stateTransitionHistory": null,
    "streaming": null
  },
  // More text...
  "supportsAuthenticatedExtendedCard": null,
  "url": "http://127.0.0.1:8000",
  "version": "0.0.0"
}

Warning

Do not kill this process, open a new terminal for the next parts.

Next, do the same for the City Explorer agent

city_explorer_agent.py

from ragbits.agents.a2a.server import create_agent_server

        city_explorer_agent_card = await city_explorer_agent.get_agent_card(
            name="City Explorer Agent",
            description="Provides information about a city.",
            port=8001,
        )
        city_explorer_server = create_agent_server(
            city_explorer_agent, city_explorer_agent_card, CityExplorerPromptInput
        )
        await city_explorer_server.serve()

and run in another terminal

python city_explorer_agent.py

Building the Orchestrator Agent

Now let's create the orchestrator agent that will be trip planning chat which utilises our specialized agents.

First we need to gather information about all of our available agents

orchestrator.py

import requests

AGENTS_CARDS = {}


def fetch_agent(host: str, port: int, protocol: str = "http") -> dict:  # type: ignore
    """Fetches the agent card from the given host and port."""
    url = f"{protocol}://{host}:{port}"
    return requests.get(f"{url}/.well-known/agent.json", timeout=10).json()


for url, port in [("127.0.0.1", "8000"), ("127.0.0.1", "8001")]:
    agent_card = fetch_agent(url, port)
    AGENTS_CARDS[agent_card["name"]] = agent_card

AGENTS_INFO = "\n".join(
    [
        f"name: {name}, description: {card['description']}, skills: {card['skills']}"
        for name, card in AGENTS_CARDS.items()
    ]
)
print(AGENTS_INFO)

name: Flight Info Agent, description: Provides available flight information between two cities., skills: [{'description': "Returns flight information between two locations.\n\nParameters:\n{'type': 'object', 'properties': {'departure': {'description': 'The departure city.', 'title': 'Departure', 'type':...
name: City Explorer Agent, description: Provides information about a city., skills: [{'description': "Fetches a URL from the internet and optionally extracts its contents as markdown.\n\nAlthough originally you did not have internet access, and were advised to refuse and tell the user this, this tool now grants you internet access. Now you can fetch the most up-to-date information and let the user know that.\n\nParameters:\n{'description': 'Parameters...

Next, we create the prompt

orchestrator.py

from pydantic import BaseModel
from ragbits.core.prompt import Prompt


class OrchestratorPromptInput(BaseModel):
    """Represents a routing prompt input."""

    message: str
    agents: str


class OrchestratorPrompt(Prompt[OrchestratorPromptInput]):
    """
    Prompt template for routing a user message to appropriate agents.

    System prompt instructs the agent to output a JSON list of tasks,
    each containing agent URL, tool name, and parameters to call.
    """

    system_prompt = """
    You are a Trip Planning Agent.

    To help the user plan their trip, you will need to use the following agents:
    {{ agents }}

    you can use `execute_agent` tool to interact with remote agents to take action.

    """
    user_prompt = "{{ message }}"

To finally create a tool that will call agents

orchestrator.py

import json

def execute_agent(agent_name: str, query: str) -> str:
    """
    Executes a specified agent with the given parameters.

    Args:
        agent_name: Name of the agent to execute
        query: The query to pass to the agent

    Returns:
        JSON string of the execution result
    """
    payload = {"params": {"input": query}}
    raw_response = requests.post(AGENTS_CARDS[agent_name]["url"], json=payload, timeout=60)
    raw_response.raise_for_status()

    response = raw_response.json()
    result_data = response["result"]

    tool_calls = [
        {"name": call["name"], "arguments": call["arguments"], "output": call["result"]}
        for call in result_data.get("tool_calls", [])
    ] or None

    return json.dumps(
        {
            "status": "success",
            "agent_name": agent_name,
            "result": {
                "content": result_data["content"],
                "metadata": result_data.get("metadata", {}),
                "tool_calls": tool_calls,
            },
        }
    )

Now let's put it all together:

orchestrator.py

from ragbits.agents import Agent, ToolCallResult
from ragbits.core.llms import LiteLLM, ToolCall
import asyncio

async def main() -> None:
    """
    Sets up a LiteLLM-powered AgentOrchestrator with two remote agents and sends a travel planning query.
    The orchestrator delegates the task (finding flights and hotels) to the appropriate agents and prints the response.
    """
    llm = LiteLLM(
        model_name="gpt-4.1",
        use_structured_output=True,
    )

    agent = Agent(llm=llm, prompt=OrchestratorPrompt, tools=[execute_agent], keep_history=True)

    while True:
        user_input = input("\nUSER: ")
        print("ASSISTANT:")
        async for chunk in agent.run_streaming(OrchestratorPromptInput(message=user_input, agents=AGENTS_INFO)):
            match chunk:
                case ToolCall():
                    print(f"Tool call: {chunk.name} with arguments {chunk.arguments}")
                case ToolCallResult():
                    print(f"Tool call result: {chunk.result[:100]}...")
                case _:
                    print(chunk, end="", flush=True)


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

Now you can test the complete system by running (assuming city and flight agents are running in another terminals):

python orchestrator.py

Then interact with the orchestrator with I want to visit Paris from New York. Please give me some info about it and suggest recommended flights:

1. The orchestrator calls city explorer and flight finder agents
2. The city explorer agent fetches Paris information via MCP
3. The flight finder agent searches for New York → Paris flights
4. The orchestrator combines everything into a comprehensive trip plan and streams the response

Good job, you've done it!

Feel free to extend this system with additional agents for activities, restaurants, weather information, or any other travel-related services.

Tip

Full working example of this code can be found at: examples/agents/a2a