Skip to main content

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

FeatureGoogle ADKClaude Agent SDKPydantic AILangGraph
Function toolstools=[my_fn]tools=[{name, description, input_schema}]@agent.tool decorator or tools=[my_fn]bind_tools([my_fn])
CRAFT MCP connectionMCPToolset + StreamableHTTPConnectionParamsmcp SDK streamablehttp_client + function toolsFastMCPToolset + StreamableHttpTransport (first-class)MultiServerMCPClient (transport="http")
Parallel tool callsYes (automatic)Yes (with betas=["interleaved-thinking"])parallel_tool_calls=True in ModelSettingsNode-level parallelism
Tool retry on failureADK handles automaticallyManual in tool loopModelRetry exceptionConditional edges
Context injectionToolContextManual in tool dispatchRunContext[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.