> ## Documentation Index
> Fetch the complete documentation index at: https://docs.emergence.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 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"

```python theme={null}
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.

<AccordionGroup>
  <Accordion title="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.

    ```python theme={null}
    from typing import Literal

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

  <Accordion title="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.

    ```python theme={null}
    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."""
        ...
    ```
  </Accordion>

  <Accordion title="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.

    ```python theme={null}
    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)}
    ```
  </Accordion>
</AccordionGroup>

## 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:

```python theme={null}
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`:

```python theme={null}
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],
)
```

<Note>
  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+).
</Note>

### 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:

```python theme={null}
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`):

```python theme={null}
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
```

<Note>
  `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](https://github.com/langchain-ai/langchain-mcp-adapters/issues/194).
</Note>

## 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.

```python theme={null}
# 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."
        ),
    }
```

<Warning>
  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.
</Warning>

## Pre-Flight Validation

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

```python theme={null}
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

<CardGroup cols={2}>
  <Card title="Multi-Agent Patterns" icon="diagram-project" href="/guides/agent-author/multi-agent-patterns">
    Use tools and sub-agents together in orchestrated workflows.
  </Card>

  <Card title="Debugging Agents" icon="bug" href="/guides/agent-author/debugging-agents">
    Inspect tool call traces and diagnose tool failures.
  </Card>
</CardGroup>