AQUAVIEW MCP with Gemini (Google Gen AI SDK)¶

This notebook shows how to give Gemini direct access to the AQUAVIEW MCP catalog using the Google Gen AI SDK's native MCP tool support — no proxy, no local server.

We'll ask Gemini the same question as in the Claude and ChatGPT notebooks:

Find Argo float profiles within 200 km of Hawaii from March 2026 where the float reached at least 1500 dbar pressure. Summarize the temperature range observed and give me a few download links.

Gemini will autonomously call AQUAVIEW's search_datasets, get_item, and possibly aggregate tools to answer.

Sources used: GADR (Global Argo Data Repository, NOAA NCEI)

Install¶

You need the google-genai SDK and the mcp Python client (used internally for the MCP transport).

In [ ]:

%pip install -U google-genai mcp

%pip install -U google-genai mcp

Setup¶

Set your Gemini API key in the GOOGLE_API_KEY environment variable. You can get one free at https://aistudio.google.com/apikey.

In [ ]:

import os
from google import genai
from google.genai import types
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

assert os.environ.get("GOOGLE_API_KEY"), "Set GOOGLE_API_KEY in your environment"

client = genai.Client()

AQUAVIEW_URL = "https://mcp.aquaview.org/mcp"
MODEL = "gemini-2.5-pro"

import os from google import genai from google.genai import types from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client assert os.environ.get("GOOGLE_API_KEY"), "Set GOOGLE_API_KEY in your environment" client = genai.Client() AQUAVIEW_URL = "https://mcp.aquaview.org/mcp" MODEL = "gemini-2.5-pro"

Connect to AQUAVIEW MCP and ask the question¶

The Gemini SDK accepts an active MCP ClientSession as a tool. Open a streamable-HTTP session to AQUAVIEW, then pass the session into tools=[...]. Gemini will call AQUAVIEW's tools autonomously.

In [ ]:

PROMPT = (
    "Find Argo float profiles within 200 km of Hawaii from March 2026 "
    "where the float reached at least 1500 dbar pressure. Summarize the "
    "temperature range observed and give me a few download links."
)

async with streamablehttp_client(AQUAVIEW_URL) as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()

        response = await client.aio.models.generate_content(
            model=MODEL,
            contents=PROMPT,
            config=types.GenerateContentConfig(
                tools=[session],
                automatic_function_calling=types.AutomaticFunctionCallingConfig(
                    maximum_remote_calls=10,
                ),
            ),
        )

print(f"finish_reason: {response.candidates[0].finish_reason}")
if response.usage_metadata:
    print(f"prompt tokens   : {response.usage_metadata.prompt_token_count}")
    print(f"candidates tokens: {response.usage_metadata.candidates_token_count}")

PROMPT = ( "Find Argo float profiles within 200 km of Hawaii from March 2026 " "where the float reached at least 1500 dbar pressure. Summarize the " "temperature range observed and give me a few download links." ) async with streamablehttp_client(AQUAVIEW_URL) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() response = await client.aio.models.generate_content( model=MODEL, contents=PROMPT, config=types.GenerateContentConfig( tools=[session], automatic_function_calling=types.AutomaticFunctionCallingConfig( maximum_remote_calls=10, ), ), ) print(f"finish_reason: {response.candidates[0].finish_reason}") if response.usage_metadata: print(f"prompt tokens : {response.usage_metadata.prompt_token_count}") print(f"candidates tokens: {response.usage_metadata.candidates_token_count}")

Inspect what Gemini did¶

When automatic_function_calling is enabled, Gemini calls the MCP tools internally and returns the final text. The full call history is available in response.automatic_function_calling_history.

In [ ]:

from textwrap import shorten

history = response.automatic_function_calling_history or []
for i, turn in enumerate(history):
    role = turn.role
    for part in (turn.parts or []):
        if part.function_call:
            name = part.function_call.name
            args = dict(part.function_call.args or {})
            print(f"[{i}/{role}] tool_call → aquaview.{name}({args})")
        elif part.function_response:
            name = part.function_response.name
            resp_data = dict(part.function_response.response or {})
            preview = shorten(
                str(resp_data).replace("\n", " "), width=240, placeholder="…"
            )
            print(f"[{i}/{role}] tool_result {name}")
            print(f"    {preview}")
        elif part.text:
            print(f"[{i}/{role}] text\n{part.text}\n")

from textwrap import shorten history = response.automatic_function_calling_history or [] for i, turn in enumerate(history): role = turn.role for part in (turn.parts or []): if part.function_call: name = part.function_call.name args = dict(part.function_call.args or {}) print(f"[{i}/{role}] tool_call → aquaview.{name}({args})") elif part.function_response: name = part.function_response.name resp_data = dict(part.function_response.response or {}) preview = shorten( str(resp_data).replace("\n", " "), width=240, placeholder="…" ) print(f"[{i}/{role}] tool_result {name}") print(f" {preview}") elif part.text: print(f"[{i}/{role}] text\n{part.text}\n")

Typical trace for this prompt:

1. tool_call → aquaview.search_datasets(collections='GADR', bbox='-161,18,-154,23', datetime='2026-03-01.../2026-03-31...', filter={...Pressure.max >= 1500...})
2. tool_result — AQUAVIEW's CSV of float records with column_stats_summary
3. tool_call → aquaview.get_item(collection='GADR', item_id='argo_jma_7900869') to fetch download URLs
4. tool_result — full STAC item with assets
5. text — Gemini's narrative summary citing the floats and their NCEI / Ifremer GDAC URLs

Render Gemini's final answer¶

In [ ]:

from IPython.display import Markdown, display

display(Markdown(response.text))

from IPython.display import Markdown, display display(Markdown(response.text))

Multi-turn — follow up¶

Re-use the same MCP session and pass the conversation history back as contents.

In [ ]:

async with streamablehttp_client(AQUAVIEW_URL) as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()

        # Reconstruct conversation: original prompt + Gemini's reply + new user turn
        contents = [
            types.Content(role="user", parts=[types.Part(text=PROMPT)]),
            types.Content(role="model", parts=[types.Part(text=response.text)]),
            types.Content(role="user", parts=[types.Part(text=(
                "For the deepest-diving float you found, give me its full lifetime "
                "trajectory bounding box and the direct NetCDF download URL."
            ))]),
        ]

        followup = await client.aio.models.generate_content(
            model=MODEL,
            contents=contents,
            config=types.GenerateContentConfig(
                tools=[session],
                automatic_function_calling=types.AutomaticFunctionCallingConfig(
                    maximum_remote_calls=10,
                ),
            ),
        )

display(Markdown(followup.text))

async with streamablehttp_client(AQUAVIEW_URL) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() # Reconstruct conversation: original prompt + Gemini's reply + new user turn contents = [ types.Content(role="user", parts=[types.Part(text=PROMPT)]), types.Content(role="model", parts=[types.Part(text=response.text)]), types.Content(role="user", parts=[types.Part(text=( "For the deepest-diving float you found, give me its full lifetime " "trajectory bounding box and the direct NetCDF download URL." ))]), ] followup = await client.aio.models.generate_content( model=MODEL, contents=contents, config=types.GenerateContentConfig( tools=[session], automatic_function_calling=types.AutomaticFunctionCallingConfig( maximum_remote_calls=10, ), ), ) display(Markdown(followup.text))

Streaming¶

For interactive UIs, stream the response.

In [ ]:

async with streamablehttp_client(AQUAVIEW_URL) as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()

        async for chunk in await client.aio.models.generate_content_stream(
            model=MODEL,
            contents=(
                "How many Argo profile observations are in the AQUAVIEW catalog "
                "from 2025? Use an aggregation, don't enumerate."
            ),
            config=types.GenerateContentConfig(
                tools=[session],
                automatic_function_calling=types.AutomaticFunctionCallingConfig(
                    maximum_remote_calls=10,
                ),
            ),
        ):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print()

async with streamablehttp_client(AQUAVIEW_URL) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() async for chunk in await client.aio.models.generate_content_stream( model=MODEL, contents=( "How many Argo profile observations are in the AQUAVIEW catalog " "from 2025? Use an aggregation, don't enumerate." ), config=types.GenerateContentConfig( tools=[session], automatic_function_calling=types.AutomaticFunctionCallingConfig( maximum_remote_calls=10, ), ), ): if chunk.text: print(chunk.text, end="", flush=True) print()

Where to go next¶

• Same notebook in Claude / ChatGPT: see 01-claude-mcp-agent.ipynb and 02-chatgpt-mcp-agent.ipynb — same prompt, three different agent runtimes.
• Other examples: see ../examples/ for 20+ more research questions, with prompts and transcripts.
• Tool reference: full parameter docs in ../docs/tools-reference.md.
• Manual function calling: if you want to inspect each tool call before passing it back to the model, set automatic_function_calling.disable=True and drive the loop yourself.