Tool Authoring

Tools are the primary way agents access external capabilities — databases, APIs, file storage, computation environments. Getting tools right is critical: poorly specified tools are the leading cause of agent errors, context bloat, and cost overruns.

Function Tools

Function tools are Python functions that the LLM can call directly. Every supported framework converts them to the appropriate wire format for the underlying model.

Writing a Good Tool

A well-authored function tool has four properties:

A descriptive docstring — the LLM reads this to decide when to call the tool
Typed parameters — every parameter has an explicit type annotation and docstring description
A bounded return type — string, dict, or a typed Pydantic model; never untyped or variable-shape
Minimal side effects — each call does one thing; avoid tools that “do everything”

def get_schema(table_name: str, schema_fqn: str) -> dict:
    """Return the column definitions for a database table.

    Call this before generating SQL to understand the table structure.
    Returns an empty dict if the table does not exist.

    Args:
        table_name: The simple table name (e.g. "orders").
        schema_fqn: The fully qualified schema name (e.g. "analytics-db.analytics_db.public").

    Returns:
        A dict with keys: columns (list of {name, type, nullable}), row_count (int).
    """
    # ... implementation
    return {"columns": [...], "row_count": 1234}

Schema Discipline

The tool’s parameter schema is sent to the LLM on every call. Verbose or inaccurate schemas degrade routing quality and inflate cost.

Use narrow types

Prefer Literal["postgres", "redshift", "bigquery"] over str when the set of valid values is known. The LLM will hallucinate less when the schema constrains choices.

from typing import Literal

def execute_query(
    sql: str,
    dialect: Literal["postgres", "redshift", "bigquery"],
) -> dict:
    """Execute a SQL query against the target database."""
    ...

Mark optional parameters explicitly

Use Optional[T] with a sensible default. Never use Union[T, None] without a default — the LLM will try to supply a value when it shouldn’t.

from typing import Optional

def search_artifacts(
    query: str,
    max_results: int = 10,
    artifact_type: Optional[Literal["parquet", "chart", "code"]] = None,
) -> list[dict]:
    """Search for artifacts matching the query."""
    ...

Return structured errors, not exceptions

Raise only for truly unrecoverable situations. For expected failures (table not found, invalid SQL), return a structured error dict so the LLM can understand and recover.

def execute_sql(sql: str) -> dict:
    """Execute SQL and return results or a structured error."""
    try:
        rows = db.execute(sql)
        return {"success": True, "rows": rows, "error": None}
    except QueryError as e:
        # Return the error — don't raise. The LLM will retry with corrected SQL.
        return {"success": False, "rows": [], "error": str(e)}

MCP Tools via FastMCPToolset

CRAFT exposes platform capabilities through em-runtime-mcp — an MCP server. Pydantic AI connects to it via FastMCPToolset; Google ADK can connect via a MCPToolset wrapper.

Pydantic AI — FastMCPToolset

A production-grade pattern for connecting a Pydantic AI agent to the CRAFT MCP server:

from pydantic_ai.toolsets.fastmcp import FastMCPToolset
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport


# Create a fresh client and toolset per agent request.
# Never reuse a long-lived client — MCP sessions expire.
mcp_client = Client(
    transport=StreamableHttpTransport(
        url="https://craft.emergence.ai/mcp",
        headers={
            "Authorization": "Bearer <token>",
            "X-Project-ID": "<your-project-id>",
        },
    )
)
toolset = FastMCPToolset(mcp_client)

# Inject into the agent run
async with agent.run_stream(
    user_message,
    deps=agent_deps,
    toolsets=[toolset],
) as run:
    async for text in run.stream_text():
        ...

# Always close the client after the run
await mcp_client.close()

Google ADK — MCP Tool Integration

Google ADK connects to the CRAFT MCP server using StreamableHTTPConnectionParams:

from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams
from google.adk.agents import LlmAgent

toolset = MCPToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://craft.emergence.ai/mcp",
        headers={
            "Authorization": "Bearer <token>",
            "X-Project-ID": "<your-project-id>",
        },
    )
)

root_agent = LlmAgent(
    model="gemini-3.5-flash",
    name="my_agent",
    description="Agent with CRAFT MCP tool access",
    instruction="Use available MCP tools to answer questions.",
    tools=[toolset],
)

The older import from google.adk.toolsets.mcp import MCPToolset is deprecated. Use from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams (ADK v1.0+).

Claude Agent SDK — MCP Tool Integration

The Anthropic API’s mcp_servers connector only supports authorization_token — it cannot pass X-Project-ID as a separate header. Use the mcp Python SDK with a session-aware client to call CRAFT tools from a standard function-tool loop:

import asyncio
import anthropic
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

CRAFT_URL = "https://craft.emergence.ai/mcp"
CRAFT_HEADERS = {
    "Authorization": "Bearer <token>",
    "X-Project-ID": "<your-project-id>",
}

async def run_craft_agent(user_question: str) -> str:
    client = anthropic.Anthropic()

    async with streamablehttp_client(CRAFT_URL, headers=CRAFT_HEADERS) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools_result = await session.list_tools()
            tools = [
                {
                    "name": t.name,
                    "description": t.description or "",
                    "input_schema": t.inputSchema,
                }
                for t in tools_result.tools
            ]

            messages = [{"role": "user", "content": user_question}]
            while True:
                response = client.messages.create(
                    model="claude-opus-4-8", max_tokens=1024,
                    tools=tools, messages=messages,
                )
                if response.stop_reason == "end_turn":
                    return response.content[0].text

                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = await session.call_tool(block.name, block.input)
                        text = result.content[0].text if result.content else ""
                        tool_results.append(
                            {"type": "tool_result", "tool_use_id": block.id, "content": text}
                        )
                messages += [
                    {"role": "assistant", "content": response.content},
                    {"role": "user", "content": tool_results},
                ]

LangGraph — MCP Tool Integration

LangGraph connects to CRAFT via MultiServerMCPClient from langchain-mcp-adapters (install: pip install langchain-mcp-adapters):

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

async def build_craft_agent():
    async with MultiServerMCPClient({
        "craft": {
            "url": "https://craft.emergence.ai/mcp",
            "transport": "http",  # streamable-HTTP (MCP 2025-11-25); "sse" is the legacy transport
            "headers": {
                "Authorization": "Bearer <token>",
                "X-Project-ID": "<your-project-id>",
            },
        }
    }) as mcp_client:
        tools = await mcp_client.get_tools()
        agent = create_react_agent("anthropic:claude-opus-4-8", tools)
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "List my data connections"}]}
        )
        return result

langchain-mcp-adapters supports headers for static auth values. If the access token expires mid-session, reinitialise the client — dynamic per-request token injection is a known limitation.

Cross-Framework Tool Comparison

Feature	Google ADK	Claude Agent SDK	Pydantic AI	LangGraph
Function tools	`tools=[my_fn]`	`tools=[{name, description, input_schema}]`	`@agent.tool` decorator or `tools=[my_fn]`	`bind_tools([my_fn])`
CRAFT MCP connection	`MCPToolset` + `StreamableHTTPConnectionParams`	`mcp` SDK `streamablehttp_client` + function tools	`FastMCPToolset` + `StreamableHttpTransport` (first-class)	`MultiServerMCPClient` (`transport="http"`)
Parallel tool calls	Yes (automatic)	Yes (with `betas=["interleaved-thinking"]`)	`parallel_tool_calls=True` in `ModelSettings`	Node-level parallelism
Tool retry on failure	ADK handles automatically	Manual in tool loop	`ModelRetry` exception	Conditional edges
Context injection	`ToolContext`	Manual in tool dispatch	`RunContext[Deps]`	`RunnableConfig`

Limiting Iteration and Cost

Unbounded tool call loops are the leading cause of cost overruns. Implement iteration limits defensively.

# Pydantic AI — track failures on deps
if not hasattr(ctx.deps, "_code_failure_count"):
    ctx.deps._code_failure_count = 0

ctx.deps._code_failure_count += 1

if ctx.deps._code_failure_count >= settings.max_code_failures:
    # Return a structured error instead of raising ModelRetry.
    # Raising ModelRetry here exhausts the retry budget and raises
    # UnexpectedModelBehavior — losing the user-facing message.
    return {
        "success": False,
        "error": (
            f"Execution has failed {ctx.deps._code_failure_count} times. "
            "Ask the user for clarification."
        ),
    }

Return structured error dicts (not raise ModelRetry) when you have hit an iteration limit. ModelRetry with the default max_retries=1 already consumed will raise UnexpectedModelBehavior and produce a generic failure message instead of the clarifying ask you wanted the model to emit.

Pre-Flight Validation

Validate tool inputs before making expensive network calls. A FastMCPToolset subclass can run a lint pass before code execution:

async def call_tool(self, name, tool_args, ctx, tool):
    if name == "run_code":
        code = tool_args.get("code", "")
        violations = lint_code(code)
        if violations:
            # Reject before the sandbox round-trip.
            # Saves latency and gives the model a sharper error message.
            ctx.deps._code_failure_count += 1
            return {
                "success": False,
                "error": f"Forbidden pattern(s): {', '.join(v.pattern for v in violations)}",
            }
    # ... proceed to actual tool call

Tool Authoring

Tool Authoring

Function Tools

Writing a Good Tool

Schema Discipline

MCP Tools via FastMCPToolset

Pydantic AI — FastMCPToolset

Google ADK — MCP Tool Integration

Claude Agent SDK — MCP Tool Integration

LangGraph — MCP Tool Integration

Cross-Framework Tool Comparison

Limiting Iteration and Cost

Pre-Flight Validation

Next Steps

Multi-Agent Patterns

Debugging Agents

​Tool Authoring

​Function Tools

​Writing a Good Tool

​Schema Discipline

​MCP Tools via FastMCPToolset

​Pydantic AI — FastMCPToolset

​Google ADK — MCP Tool Integration

​Claude Agent SDK — MCP Tool Integration

​LangGraph — MCP Tool Integration

​Cross-Framework Tool Comparison

​Limiting Iteration and Cost

​Pre-Flight Validation

​Next Steps

Multi-Agent Patterns

Debugging Agents

Tool Authoring

Function Tools

Writing a Good Tool

Schema Discipline

MCP Tools via FastMCPToolset

Pydantic AI — FastMCPToolset

Google ADK — MCP Tool Integration

Claude Agent SDK — MCP Tool Integration

LangGraph — MCP Tool Integration

Cross-Framework Tool Comparison

Limiting Iteration and Cost

Pre-Flight Validation

Next Steps