AQUAVIEW MCP with Claude (Anthropic Python SDK)¶

This notebook shows how to give Claude direct access to the AQUAVIEW MCP catalog using the Anthropic API's native mcp_servers parameter — no proxy, no local server, no shim.

We'll ask Claude:

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.

Claude 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 only need the anthropic SDK. AQUAVIEW is hosted, no other dependencies.

In [ ]:

%pip install -U anthropic

%pip install -U anthropic

Setup¶

Set your Anthropic API key in the ANTHROPIC_API_KEY environment variable before running this notebook (or replace os.environ[...] below with your key inline — but don't commit it).

In [ ]:

import os
from anthropic import Anthropic

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

client = Anthropic()

AQUAVIEW = {
    "type": "url",
    "url": "https://mcp.aquaview.org/mcp",
    "name": "aquaview",
}

MODEL = "claude-opus-4-7"

import os from anthropic import Anthropic assert os.environ.get("ANTHROPIC_API_KEY"), "Set ANTHROPIC_API_KEY in your environment" client = Anthropic() AQUAVIEW = { "type": "url", "url": "https://mcp.aquaview.org/mcp", "name": "aquaview", } MODEL = "claude-opus-4-7"

Ask the question¶

We pass mcp_servers=[AQUAVIEW] and the model can call AQUAVIEW's tools directly. Claude will issue tool calls, the API will execute them against mcp.aquaview.org/mcp, and the model will integrate the results into its final answer — all in one messages.create call.

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

response = client.messages.create(
    model=MODEL,
    max_tokens=4096,
    mcp_servers=[AQUAVIEW],
    messages=[{"role": "user", "content": PROMPT}],
)

print(f"stop_reason: {response.stop_reason}")
print(f"input tokens : {response.usage.input_tokens}")
print(f"output tokens: {response.usage.output_tokens}")

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." ) response = client.messages.create( model=MODEL, max_tokens=4096, mcp_servers=[AQUAVIEW], messages=[{"role": "user", "content": PROMPT}], ) print(f"stop_reason: {response.stop_reason}") print(f"input tokens : {response.usage.input_tokens}") print(f"output tokens: {response.usage.output_tokens}")

Inspect what Claude did¶

The response's content field is a list of blocks. With MCP enabled you'll see three block types interleaved:

• text — Claude's reasoning / final answer
• mcp_tool_use — a tool call Claude made (which AQUAVIEW tool, with what args)
• mcp_tool_result — the data AQUAVIEW returned

In [ ]:

from textwrap import shorten

for i, block in enumerate(response.content):
    bt = getattr(block, "type", type(block).__name__)
    if bt == "text":
        print(f"[{i}] text\n{block.text}\n")
    elif bt == "mcp_tool_use":
        args = getattr(block, "input", {}) or {}
        print(f"[{i}] tool_use → aquaview.{block.name}({args})")
    elif bt == "mcp_tool_result":
        first = block.content[0].text if block.content else ""
        preview = shorten(first.replace("\n", " "), width=240, placeholder="…")
        print(f"[{i}] tool_result\n    {preview}\n")
    else:
        print(f"[{i}] {bt}")

from textwrap import shorten for i, block in enumerate(response.content): bt = getattr(block, "type", type(block).__name__) if bt == "text": print(f"[{i}] text\n{block.text}\n") elif bt == "mcp_tool_use": args = getattr(block, "input", {}) or {} print(f"[{i}] tool_use → aquaview.{block.name}({args})") elif bt == "mcp_tool_result": first = block.content[0].text if block.content else "" preview = shorten(first.replace("\n", " "), width=240, placeholder="…") print(f"[{i}] tool_result\n {preview}\n") else: print(f"[{i}] {bt}")

Typical trace for this prompt:

1. mcp_tool_use → aquaview.search_datasets(collections='GADR', bbox='-161,18,-154,23', datetime='2026-03-01T00:00:00Z/2026-03-31T23:59:59Z', filter={...Pressure.max >= 1500...})
2. mcp_tool_result — sectioned CSV of float records with column_stats_summary
3. mcp_tool_use → aquaview.get_item(collection='GADR', item_id='argo_jma_7900869') — to fetch download URLs for the top result
4. mcp_tool_result — full STAC item with assets
5. text — Claude's narrative summary citing the floats and their NCEI / Ifremer GDAC download URLs

Render Claude's final answer¶

Pull just the text blocks for a clean read.

In [ ]:

from IPython.display import Markdown, display

answer = "\n\n".join(b.text for b in response.content if getattr(b, "type", "") == "text")
display(Markdown(answer))

from IPython.display import Markdown, display answer = "\n\n".join(b.text for b in response.content if getattr(b, "type", "") == "text") display(Markdown(answer))

Extract structured data¶

The mcp_tool_result blocks contain AQUAVIEW's raw responses. AQUAVIEW defaults to a sectioned CSV that's compact for the model, but for downstream code you'll usually want JSON. We can ask Claude to re-issue queries with output_format='json' — or parse the CSV ourselves.

Here's a quick recipe to pull the search-result IDs out of the trace and then do a programmatic, deterministic follow-up using the same MCP server (no LLM in the loop) via the mcp Python client:

In [ ]:

# Pull item IDs out of Claude's tool-result CSV blocks
import csv
from io import StringIO

ids = []
for block in response.content:
    if getattr(block, "type", "") != "mcp_tool_result":
        continue
    text = block.content[0].text if block.content else ""
    # AQUAVIEW CSV uses '## items' as the section header
    if "## items" not in text:
        continue
    body = text.split("## items", 1)[1]
    # Skip the '# rows: N' line and the header line
    lines = [l for l in body.splitlines() if l and not l.startswith("#")]
    if len(lines) < 2:
        continue
    reader = csv.DictReader(StringIO("\n".join(lines)))
    for row in reader:
        ids.append(row.get("id"))

ids = [i for i in ids if i]
ids[:10]

# Pull item IDs out of Claude's tool-result CSV blocks import csv from io import StringIO ids = [] for block in response.content: if getattr(block, "type", "") != "mcp_tool_result": continue text = block.content[0].text if block.content else "" # AQUAVIEW CSV uses '## items' as the section header if "## items" not in text: continue body = text.split("## items", 1)[1] # Skip the '# rows: N' line and the header line lines = [l for l in body.splitlines() if l and not l.startswith("#")] if len(lines) < 2: continue reader = csv.DictReader(StringIO("\n".join(lines))) for row in reader: ids.append(row.get("id")) ids = [i for i in ids if i] ids[:10]

Multi-turn — follow up on a specific float¶

Claude can keep working with AQUAVIEW across turns. Pass the prior messages list back along with a new user message; Claude may issue more tool calls.

In [ ]:

messages = [
    {"role": "user", "content": PROMPT},
    {"role": "assistant", "content": response.content},
    {"role": "user", "content":
        "For the deepest-diving float you found, give me its full lifetime "
        "trajectory bounding box and the direct NetCDF download URL."
    },
]

followup = client.messages.create(
    model=MODEL,
    max_tokens=2048,
    mcp_servers=[AQUAVIEW],
    messages=messages,
)

display(Markdown(
    "\n\n".join(b.text for b in followup.content if getattr(b, "type", "") == "text")
))

messages = [ {"role": "user", "content": PROMPT}, {"role": "assistant", "content": response.content}, {"role": "user", "content": "For the deepest-diving float you found, give me its full lifetime " "trajectory bounding box and the direct NetCDF download URL." }, ] followup = client.messages.create( model=MODEL, max_tokens=2048, mcp_servers=[AQUAVIEW], messages=messages, ) display(Markdown( "\n\n".join(b.text for b in followup.content if getattr(b, "type", "") == "text") ))

Streaming¶

For interactive UIs, stream the response so you can render Claude's reasoning + tool calls + tool results as they arrive.

In [ ]:

with client.messages.stream(
    model=MODEL,
    max_tokens=2048,
    mcp_servers=[AQUAVIEW],
    messages=[{"role": "user", "content":
        "How many Argo profile observations are in the AQUAVIEW catalog "
        "from 2025? Use an aggregation, don't enumerate."
    }],
) as stream:
    for event in stream:
        et = event.type
        if et == "content_block_start":
            block = event.content_block
            if block.type == "mcp_tool_use":
                print(f"[tool] aquaview.{block.name}")
        elif et == "text":
            print(event.text, end="", flush=True)
    print()

with client.messages.stream( model=MODEL, max_tokens=2048, mcp_servers=[AQUAVIEW], messages=[{"role": "user", "content": "How many Argo profile observations are in the AQUAVIEW catalog " "from 2025? Use an aggregation, don't enumerate." }], ) as stream: for event in stream: et = event.type if et == "content_block_start": block = event.content_block if block.type == "mcp_tool_use": print(f"[tool] aquaview.{block.name}") elif et == "text": print(event.text, end="", flush=True) print()

Where to go next¶

• Other examples: see ../examples/ for 20+ more research questions, with prompts and transcripts.
• Tool reference: full parameter docs in ../docs/tools-reference.md.
• Same notebook in ChatGPT / Gemini: see 02-chatgpt-mcp-agent.ipynb and 03-gemini-mcp-agent.ipynb (coming soon) — same prompt, three different agent runtimes.
• Build a real app: this mcp_servers pattern works inside any production Anthropic API call. Combine with extended thinking and structured output for research-grade pipelines.