Tool Authoring

How to write function tools, connect MCP tools via FastMCPToolset, and apply schema discipline across frameworks.

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:

1. A descriptive docstring — the LLM reads this to decide when to call the tool
2. Typed parameters — every parameter has an explicit type annotation and docstring description
3. A bounded return type — string, dict, or a typed Pydantic model; never untyped or variable-shape
4. 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-2.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-sonnet-4-6", 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-sonnet-4-6", 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

Next Steps

Multi-Agent Patterns

Use tools and sub-agents together in orchestrated workflows.

Debugging Agents

Inspect tool call traces and diagnose tool failures.