Claude Agent SDK

The Claude Agent SDK integration ships in the Python and TypeScript SDKs. It has two halves:

1. The Marmot MCP server (built into your Marmot instance at /api/v1/mcp) gives the agent catalog tools such as discover_data, find_ownership, lookup_term and trace_lineage. Point the SDK at it with mcpServers and those tools appear alongside the agent's own.
2. MarmotAgentTracker plugs into the SDK's hook system. The first time any hook fires it registers the agent as an asset of type Agent, captures every tool output for mrn:// references and writes one batched lineage call when the session ends.

Install

pip install "marmot-sdk[claude-agent]"

The claude-agent extra adds claude-agent-sdk.

pnpm add @marmotdata/sdk @anthropic-ai/claude-agent-sdk

@anthropic-ai/claude-agent-sdk is loaded only when you import the tracker, so the SDK stays lean for non-agent users.

Quick start

A minimal agent that searches the catalog, registers itself and writes lineage:

import asyncio
import sys

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, ResultMessage, TextBlock

from marmot import AuthenticatedApiClient, mcp_url
from marmot.auth import SecurityScheme, resolve_credential, resolve_host
from marmot.integrations import MarmotCatalog
from marmot.integrations.claude_agent import MarmotAgentTracker

SYSTEM_PROMPT = (
    "You answer questions about an organisation's data using the Marmot catalog. "
    "Always consult the marmot tools; never guess from memory."
)


async def main() -> None:
    host = resolve_host()
    credential = resolve_credential(host)
    client = AuthenticatedApiClient(host, credential)
    tracker = MarmotAgentTracker(
        MarmotCatalog(client),
        name="catalog-explorer",
        model="claude-sonnet-4-5",
        owner="data-eng",
    )

    token = credential.get_token()
    mcp_headers = (
        {"Authorization": f"Bearer {token}"}
        if credential.scheme is SecurityScheme.bearer
        else {"X-API-Key": token}
    )

    options = ClaudeAgentOptions(
        mcp_servers={
            "marmot": {
                "type": "http",
                "url": mcp_url(client),
                "headers": mcp_headers,
            }
        },
        hooks=tracker.hooks(),
        permission_mode="bypassPermissions",
        # Without this the agent loads your own settings and CLAUDE.md files, and
        # answers catalog questions from that context instead of asking Marmot.
        setting_sources=[],
        system_prompt=SYSTEM_PROMPT,
    )

    async with ClaudeSDKClient(options=options) as agent:
        await agent.query("Find a postgres table about orders and summarise it.")
        async for message in agent.receive_response():
            for block in getattr(message, "content", None) or []:
                if isinstance(block, TextBlock):
                    print(block.text)
            if isinstance(message, ResultMessage) and message.is_error:
                print(f"agent error: {message.errors or message.result}", file=sys.stderr)

    print("agent registered as:", tracker.agent_mrn or "(not yet registered)")


asyncio.run(main())

import { Client, resolve } from "@marmotdata/sdk";
import { MarmotAgentTracker } from "@marmotdata/sdk/claude-agent";
import { query } from "@anthropic-ai/claude-agent-sdk";

const { baseUrl, credential } = await resolve({});
const client = new Client({ baseUrl, credential });

const tracker = new MarmotAgentTracker(client, {
  name: "catalog-explorer",
  model: "claude-sonnet-4-5",
  owner: "data-eng",
});

for await (const msg of query({
  prompt: "Find a postgres table about orders and summarise it.",
  options: {
    mcpServers: {
      marmot: {
        type: "http",
        url: `${baseUrl}/api/v1/mcp`,
        headers: { Authorization: `Bearer ${credential.token}` },
      },
    },
    hooks: tracker.hooks(),
    permissionMode: "bypassPermissions",
  },
})) {
  // stream messages as you wish
}

console.log("agent registered as:", tracker.agentMrn);

After the first run the agent appears in Marmot as type=Agent, service=ClaudeAgent, name=catalog-explorer, with lineage edges from every asset it touched.

Catalog tools (via MCP)

The Marmot MCP server exposes the catalog as tools, namespaced mcp__marmot__*:

Tool	Purpose
discover_data	Find or browse assets by name, type, provider, tags or MRN
find_ownership	Resolve who owns an asset, or what a team/user owns
lookup_term	Search the business glossary for term definitions

Tool responses include MRNs (often inside markdown content blocks). The tracker walks both structured objects and free text for mrn:// URIs, so lineage is captured automatically.

To restrict the agent to a subset:

options = ClaudeAgentOptions(
    mcp_servers={"marmot": {...}},
    hooks=tracker.hooks(),
    setting_sources=[],
    system_prompt=SYSTEM_PROMPT,
    allowed_tools=["mcp__marmot__discover_data", "mcp__marmot__find_ownership"],
)

options: {
  mcpServers: { marmot: { ... } },
  hooks: tracker.hooks(),
  allowedTools: ["mcp__marmot__discover_data", "mcp__marmot__find_ownership"],
}

Custom tools

Two ways to attribute lineage from tools you ship as your own MCP server alongside Marmot's.

MRNs in tool output

If your tool returns objects with mrn fields ({ mrn, ... } or { results: [{ mrn, ... }] }), or text bodies that mention mrn://... URIs, the tracker picks them up on every PostToolUse hook. This is the same mechanism that captures MRNs from Marmot's MCP responses.

Manual record_source

Use this when the upstream is only known at runtime, for example a tool that picks one of several tables:

def query_table(table: str, sql: str) -> list[dict]:
    tracker.record_source(f"mrn://table/postgres/{table}")
    return run_sql(sql)

function queryTable(table: string, sql: string) {
  tracker.recordSource(`mrn://table/postgres/${table}`);
  return runSql(sql);
}

Next steps

• MCP server: the tools and permissions behind mcpServers.
• LangChain: the same lineage capture for LangChain agents.