A2A Financial Trading Agents - Google ADK / MCP / Llama

Two specialist agents analyze a stock from opposite directions and an orchestrator weighs their cases. What makes it more than a routing example is that the two specialists are built on different frameworks: the Bear agent on Pydantic AI, the Bull agent on Google’s Agent Development Kit (ADK). The orchestrator never learns which is which. It discovers both through their A2A agent cards and calls them as tools, so the framework choice stays behind the protocol. The Agent-to-Agent (A2A) protocol is an open standard for agents to communicate and collaborate regardless of the framework or vendor they were built with, covering agent discovery through agent cards, asynchronous task execution, structured message passing, and transport negotiation. Each agent’s own tools come from the Model Context Protocol (MCP), running as a subprocess the agent spawns over stdio. This guide covers what the system is made of, how to run it, and what its traces look like in Arize AX, including two properties of A2A tracing that will otherwise surprise you.

Project code

The runnable project this guide walks through

A2A Documentation

The protocol specification

Architecture

Component	Framework	MCP tools	Role
Bear Risk Analyst	Pydantic AI	`risk_scanner`, `divergence_detector`, `exit_signal_monitor`	Downside catalysts and warning signals
Bull Market Analyst	Google ADK	`find_breakout_patterns`, `momentum_screener`, `entry_signal_detector`	Growth opportunities and bullish patterns
Orchestrator	Google ADK	The two agents above, as A2A tools	Coordinates both and weighs the cases

Each specialist runs as an A2A service that publishes an agent card at /.well-known/agent-card.json. Market data is synthetic, so the tools need no market data feed.

orchestrator.py
  |
  |-- A2A --> localhost:8001  Bear (Pydantic AI)  --stdio--> mcp_tools/bear_mcp_server.py
  |
  '-- A2A --> localhost:8002  Bull (Google ADK)   --stdio--> mcp_tools/bull_mcp_server.py

Before you start

Clone the project and install its dependencies. Python 3.10 or later.

git clone https://github.com/Arize-ai/tutorials.git
cd tutorials/python/llm/agents/a2a_trading_agents
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

You also need an Arize AX account and access to the two Vertex AI models: Gemini 2.5 Flash for the Bear agent and orchestrator, Llama 3.3 70B for the Bull agent. On the Google Cloud side that means:

A Google Cloud project with billing and the Vertex AI API enabled
Application Default Credentials, from gcloud auth application-default login
Llama 3.3 accepted in Model Garden, which is a separate per-model license step
For Agent Engine deployment only: a GCS staging bucket and permission to create Agent Engine resources

gcloud auth application-default login

export GOOGLE_CLOUD_PROJECT="<your-project-id>"
export GOOGLE_CLOUD_LOCATION=us-central1

export ARIZE_SPACE_ID="<your-space-id>"
export ARIZE_API_KEY="<your-api-key>"
export ARIZE_PROJECT_NAME="a2a-trading-agents"

Each agent’s model id is read from the environment (BEAR_MODEL, BULL_MODEL, ORCHESTRATOR_MODEL), so you can point one at a different Vertex model without touching the agent code.

Trace both frameworks into one project

Two frameworks need two kinds of instrumentation, and both write into a single tracer provider so the orchestrator and the specialists land in the same Arize AX project. ADK is instrumented directly. Pydantic AI emits OpenTelemetry GenAI spans, so OpenInferenceSpanProcessor translates them into attributes Arize AX reads as LLM spans. Custom span processors go in register(span_processors=[...]), which keeps them alongside the exporter Arize AX sets up for you.

tracing.py

from arize.otel import register
from openinference.instrumentation.google_adk import GoogleADKInstrumentor
from openinference.instrumentation.pydantic_ai import OpenInferenceSpanProcessor
from pydantic_ai import Agent, InstrumentationSettings

tracer_provider = register(
    space_id=os.environ["ARIZE_SPACE_ID"],
    api_key=os.environ["ARIZE_API_KEY"],
    project_name=os.environ.get("ARIZE_PROJECT_NAME", "a2a-trading-agents"),
    span_processors=[OpenInferenceSpanProcessor()],  # translate Pydantic AI GenAI spans
    set_global_tracer_provider=True,
)

GoogleADKInstrumentor().instrument(tracer_provider=tracer_provider)

# Pydantic AI 2.x sets instrumentation on the Agent class; Agent(instrument=True) was removed.
Agent.instrument_all(InstrumentationSettings(tracer_provider=tracer_provider))

Give each agent its MCP tools

Both agents get their tools from an MCP server they spawn over stdio, and the two frameworks express that differently. Pydantic AI takes a transport, ADK takes connection params. Launching with sys.executable rather than "python" keeps the server in the same interpreter as the agent, which matters inside a virtualenv where python may not exist.

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset, StdioTransport

toolset = MCPToolset(
    StdioTransport(
        command=sys.executable,
        args=["-m", "mcp_tools.bear_mcp_server"],
        cwd=str(PROJECT_ROOT),
    )
)

bear_agent = Agent(
    model=config.bear_model(),
    system_prompt=BEAR_SYSTEM_PROMPT,
    toolsets=[toolset],
    retries=3,
)

from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import StdioConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

toolset = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command=sys.executable,
            args=["-m", "mcp_tools.bull_mcp_server"],
            cwd=str(PROJECT_ROOT),
        ),
        timeout=60,
    ),
)

bull_agent = LlmAgent(
    name="bull_market_analyst",
    model=config.bull_model(),
    instruction=BULL_SYSTEM_PROMPT,
    tools=[toolset],
)

Advertise each agent with an agent card

An agent card is how one agent tells another what it can do, without either knowing the other’s implementation. Skills carry examples, which is what lets the orchestrator decide when a specialist is worth calling.

from a2a.types import AgentSkill
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card

BEAR_SKILLS = [
    AgentSkill(
        id="risk_analysis",
        name="Risk Factor Scanner",
        description="Identifies potential downside catalysts and risk factors",
        tags=["Risk-Analysis", "Market-Analysis"],
        examples=[
            "What are the key risks for NVDA?",
            "Analyze downside catalysts for tech stocks",
        ],
    ),
    # divergence_detection and exit_signals follow the same shape
]


def create_bear_agent_card():
    return create_agent_card(
        agent_name="Bear Risk Analyst (Pydantic AI + MCP)",
        description=(
            "A cautious risk analyst powered by Pydantic AI, "
            "focused on identifying downside catalysts and warning signals."
        ),
        skills=BEAR_SKILLS,
    )

Bridge Pydantic AI to A2A

ADK ships an A2A executor, so the Bull agent needs no bridge. Pydantic AI does not, so the Bear agent supplies one: BearAgentExecutor turns an A2A task into an agent run and reports progress back through the TaskUpdater. Two details make it behave under failure. The agent is built lazily, because Agent Engine pickles the executor to deploy it and an initialized agent holding an MCP subprocess is not picklable. And the build happens inside the try, so a failure (bad credentials, a model the project cannot access) is reported to the caller as a failed task instead of escaping as a server-level JSON-RPC error the orchestrator cannot interpret.

class BearAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = None
        self._traced = False

    def _init_agent(self):
        if not self._traced:
            tracing.setup_tracing()  # instrument the process that serves traffic
            self._traced = True
        if self.agent is None:
            self.agent = build_bear_agent()

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        query = context.get_user_input()
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)

        if not getattr(context, "current_task", None):
            await updater.submit()
        await updater.start_work()

        try:
            # Inside the try, so a build failure becomes a failed task rather than a
            # server error the caller cannot interpret.
            self._init_agent()

            await updater.update_status(
                TaskState.working,
                message=new_agent_text_message("Analyzing risks..."),
            )

            result = await self.agent.run(query)
            result_text = getattr(result, "output", None) or str(result)

            await updater.add_artifact([TextPart(text=result_text)], name="risk_analysis")
            await updater.complete()

        except Exception as exc:
            await updater.update_status(
                TaskState.failed,
                message=new_agent_text_message(f"Analysis failed: {exc}"),
                final=True,
            )

Call the specialists over A2A

The orchestrator imports neither specialist. RemoteA2aAgent fetches each agent card and AgentTool presents the remote agent as a tool, which is the whole substitutability argument: swap either specialist’s framework, or move it to another host, and this code does not change.

orchestrator.py

from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.tools.agent_tool import AgentTool

remote_bear = RemoteA2aAgent(
    name="bear_risk_analyst",
    description="Analyzes downside risks and warning signals for a stock",
    agent_card=f"http://localhost:{config.BEAR_PORT}{AGENT_CARD_WELL_KNOWN_PATH}",
)
remote_bull = RemoteA2aAgent(
    name="bull_market_analyst",
    description="Identifies growth opportunities and bullish patterns for a stock",
    agent_card=f"http://localhost:{config.BULL_PORT}{AGENT_CARD_WELL_KNOWN_PATH}",
)

orchestrator = LlmAgent(
    name="trading_strategy_orchestrator",
    model=config.orchestrator_model(),
    instruction=ORCHESTRATOR_INSTRUCTION,
    tools=[AgentTool(agent=remote_bear), AgentTool(agent=remote_bull)],
)

Run it

One command starts both agents and sends a single query:

python run_local.py "Should I buy NVDA stock?"

Or keep the agents up across several queries, in two terminals:

python a2a_servers.py                                  # terminal 1
python orchestrator.py "What are the risks for TSLA?"  # terminal 2

The answer argues both sides from the tools’ output, a risk score and stop-loss levels from the Bear agent alongside breakout targets and an entry price from the Bull agent, then states which case is stronger.

Read the traces in Arize AX

Open the a2a-trading-agents project. One query produces roughly 100 spans:

CHAIN invocation [trading_strategy_orchestrator], the orchestrator run
AGENT and LLM spans for each agent’s reasoning, carrying the model name, so you can confirm the Bear agent ran on Gemini while the Bull agent ran on Llama
TOOL spans for the A2A calls (execute_tool bear_risk_analyst) and for every MCP tool the specialists invoke (execute_tool risk_scanner, tools/call risk_scanner)

Two properties of this trace shape are worth knowing before you go looking for them. The specialists’ work lands in separate traces from the orchestrator’s. A2A does not propagate trace context across the HTTP hop, so one query yields one orchestrator trace plus one trace per agent that answered, rather than a single connected tree. Group them by time or by project rather than expecting one root span to span the exchange. The a2a-sdk emits its own internal spans. Event-queue plumbing (EventQueue.dequeue_event and friends) accounts for most of the span count and carries no OpenInference span kind, so those rows sit uncategorized. Filter on attributes.openinference.span.kind to get to agent behavior.

Deploy to Vertex AI Agent Engine

deploy_agent_engine.py turns each specialist into a managed service with an authenticated A2A endpoint, and the orchestrator reaches them through Google-signed requests instead of localhost. It needs the GCS staging bucket and Agent Engine permissions listed above.

python deploy_agent_engine.py                                    # deploy both
python deploy_agent_engine.py --query "Analyze risks for TSLA"   # deploy, then query
python deploy_agent_engine.py --delete <resource> <resource>     # tear down

Deployment forwards your Arize AX settings to the deployed services as environment variables, so the remote agents trace into the same project. It takes several minutes per agent and leaves billable resources running, so delete them when you are finished.

Pass credentials to deployed agents through env_vars, read from your own environment. Hardcoding an API key in an executor means it ships to every deployment and into version control.

Takeaway

A2A makes the framework an implementation detail. The orchestrator calls a Pydantic AI agent and an ADK agent the same way, through cards and tasks, and neither specialist knows what the other is.
Instrument per framework, export to one project. ADK is instrumented directly and Pydantic AI through a translating span processor, but both write into a single tracer provider, so one project shows the whole system.
Trace context stops at the A2A boundary. Plan on correlating traces across agents rather than reading one tree, which is the main thing that separates multi-agent observability from single-process observability.

Instrument

Observe

Evaluate

Improve

Advanced Workflows

A2A Financial Trading Agents - Google ADK / MCP / Llama

Project code

A2A Documentation

Architecture

Before you start

Trace both frameworks into one project

Give each agent its MCP tools

Advertise each agent with an agent card

Bridge Pydantic AI to A2A

Call the specialists over A2A

Run it

Read the traces in Arize AX

Deploy to Vertex AI Agent Engine

Takeaway

Project code

A2A Documentation

​Architecture

​Before you start

​Trace both frameworks into one project

​Give each agent its MCP tools

​Advertise each agent with an agent card

​Bridge Pydantic AI to A2A

​Call the specialists over A2A

​Run it

​Read the traces in Arize AX

​Deploy to Vertex AI Agent Engine

​Takeaway

Architecture

Before you start

Trace both frameworks into one project

Give each agent its MCP tools

Advertise each agent with an agent card

Bridge Pydantic AI to A2A

Call the specialists over A2A

Run it

Read the traces in Arize AX

Deploy to Vertex AI Agent Engine

Takeaway