Skip to content

Create an Agent with a Tool¤

This guide shows how to give an agent tools using ToolSchema, Capability, and ToolRegistry. Tools let agents query databases, call APIs, or perform calculations beyond what an LLM can do on its own.

Concepts covered
  • ToolSchema — JSON Schema describing a tool's interface
  • Capability — metadata wrapper around a ToolSchema
  • ToolRegistry — scoped storage and retrieval of tools
  • Operation — with kind="artifact" and a tool-scoped agent
  • AgentConfig — with default_tool_scope to wire tools to agents
  • PydanticAIDriver — real LLM driver (or FakeDriver for tests)

What you'll build¤

A domain name generator agent that uses a word-lookup tool to find relevant keywords for a business, then combines them into creative domain name suggestions. The agent produces a typed DomainSuggestions output.

1 Define the output model¤

Every operation produces a typed output. Start with a Pydantic model:

from pydantic import BaseModel, Field


class DomainSuggestion(BaseModel):
    """A single domain name suggestion with reasoning."""

    domain: str = Field(description="The suggested domain name")
    tld: str = Field(description="Top-level domain, e.g. .com, .io")
    rationale: str = Field(description="Why this domain fits the brand")


class DomainSuggestions(BaseModel):
    """Collection of domain name suggestions."""

    brand: str = Field(description="The brand name that was analyzed")
    suggestions: list[DomainSuggestion] = Field(
        description="List of domain suggestions"
    )

2 Define tool schemas¤

A ToolSchema describes a tool in terms of its name, human-readable description, and JSON Schema for input parameters. This is the contract the agent uses to decide when and how to call the tool.

from crewmaster.tools.models import ToolSchema

WORD_LOOKUP_TOOL = ToolSchema(
    name="word_lookup",
    description="Look up related keywords and synonyms for a given term",
    input_schema={
        "type": "object",
        "properties": {
            "term": {
                "type": "string",
                "description": "The term to find related words for",
            },
            "max_results": {
                "type": "integer",
                "description": "Maximum number of results (default 5)",
                "default": 5,
            },
        },
        "required": ["term"],
    },
)

DOMAIN_CHECK_TOOL = ToolSchema(
    name="domain_check",
    description="Check availability and pricing for a domain name",
    input_schema={
        "type": "object",
        "properties": {
            "domain": {
                "type": "string",
                "description": "Domain name without TLD, e.g. 'mybrand'",
            },
            "tld": {
                "type": "string",
                "description": "TLD to check, e.g. '.com', '.io'",
                "default": ".com",
            },
        },
        "required": ["domain"],
    },
)

3 Register tools in a ToolRegistry¤

A ToolRegistry stores tools scoped by namespace. An agent uses its default_tool_scope to discover which tools it has access to.

Register each tool under a scope (here "domain_naming"), and also in the global "*" scope if you want it available everywhere:

from crewmaster.tools.registry import ToolRegistry

tool_registry = ToolRegistry()
tool_registry.register("domain_naming", WORD_LOOKUP_TOOL)
tool_registry.register("domain_naming", DOMAIN_CHECK_TOOL)
# Also register globally so any agent can discover them
tool_registry.register("*", WORD_LOOKUP_TOOL)
tool_registry.register("*", DOMAIN_CHECK_TOOL)
ToolSchema vs. Capability

Use Capability when you need richer metadata (for example, for the request_capability SAD vocabulary bridge). Capabilities wrap a ToolSchema with an extra description field for semantic matching:

from crewmaster.tools.models import Capability

tool_registry.register(
    "domain_naming",
    Capability(
        name="word_lookup",
        description="Retrieve context-aware synonyms and keywords",
        tool_schema=WORD_LOOKUP_TOOL,
    ),
)

For most cases, registering a plain ToolSchema is sufficient.

4 Create an agent with a tool scope¤

The AgentConfig sets default_tool_scope to "domain_naming". This tells the runtime to query the ToolRegistry for tools under that scope when the agent needs them:

from crewmaster.agents.agent import AgentConfig

agent = AgentConfig(
    name="domain_namer",
    identity_blocks=["blocks://domain_namer/identity"],
    default_tool_scope="domain_naming",
)

The runtime will call tool_registry.retrieve("domain_naming") to get the list of available tools, then pass them to the LLM as function definitions.

5 Define the operation¤

The Operation ties the agent, its output type, and task prompt together:

from crewmaster.operations.operation import Operation

operation = Operation(
    name="generate_domains",
    produces=DomainSuggestions,
    kind="artifact",
    agent=agent,
    task_blocks=["blocks://domain_namer/task"],
)

6 Set up prompt blocks¤

Create the identity and task templates using a LocalDiskStore:

import tempfile
from pathlib import Path

from crewmaster.agents.prompts.local_disk_store import LocalDiskStore

blocks_dir = Path(tempfile.mkdtemp())

(blocks_dir / "domain_namer" / "identity.j2").mkdir(parents=True)
(blocks_dir / "domain_namer" / "identity.j2").write_text("""\
---
id: blocks://domain_namer/identity
kind: identity
provides: domain_namer_persona
---
You are a creative brand naming expert. You use tools to research
keywords and check domain availability before making suggestions.
Always explain why each domain name fits the brand.
""")

(blocks_dir / "domain_namer" / "task.j2").write_text("""\
---
id: blocks://domain_namer/task
kind: task
provides: domain_suggestions_output
---
Generate 3 creative domain name suggestions for the brand "{{ brand_name }}"
in the {{ industry }} industry.

Steps:
1. Use word_lookup to find keywords related to "{{ brand_name }}" and "{{ industry }}"
2. Brainstorm domain names combining those keywords
3. Use domain_check to verify availability for your top picks
4. Return the suggestions as structured output

Be creative — short, memorable names are best.
""")

7 Set up the context and execute¤

Wire the ContextStore, PromptEngine, and ToolRegistry together:

import asyncio

from crewmaster import execute
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.engine import PromptEngine
from crewmaster.execution.runtime import RuntimeRequest, RuntimeResponse


# ContextStore with an inline FakeDriver for testing
store = ContextStore()
block_store = LocalDiskStore(root=str(blocks_dir))


class FakeDriver:
    """Fake driver returning a pre-canned response for testing."""

    def __init__(self, output: DomainSuggestions) -> None:
        self._output = output
        self.calls: list[RuntimeRequest] = []

    async def execute(self, request: RuntimeRequest) -> RuntimeResponse:
        self.calls.append(request)
        return RuntimeResponse(output=self._output)

    async def astream(self, request: RuntimeRequest):
        self.calls.append(request)
        from crewmaster.execution.runtime import RuntimeStreamChunk

        yield RuntimeStreamChunk(kind="text_delta", content="Researching...")
        yield RuntimeStreamChunk(kind="final", output=self._output)


async def main() -> None:
    driver = FakeDriver(
        output=DomainSuggestions(
            brand="NovaTech",
            suggestions=[
                DomainSuggestion(
                    domain="novatech.io",
                    tld=".io",
                    rationale="Short, tech-forward, matches the brand name directly",
                ),
                DomainSuggestion(
                    domain="novadev.com",
                    tld=".com",
                    rationale="Combines brand with 'dev' to signal developer tools focus",
                ),
                DomainSuggestion(
                    domain="ntstack.com",
                    tld=".com",
                    rationale="Abbreviated, professional, 'stack' suggests platform",
                ),
            ],
        )
    )

    result = await execute(
        operation=operation,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        tool_registry=tool_registry,
        prompt_engine=PromptEngine(block_store=block_store),
        context_overrides={
            "brand_name": "NovaTech",
            "industry": "developer tools",
        },
    )

    for suggestion in result.suggestions:
        print(f"  {suggestion.domain} ({suggestion.tld}) — {suggestion.rationale}")

asyncio.run(main())

Expected output:

  novatech.io (.io) — Short, tech-forward, matches the brand name directly
  novadev.com (.com) — Combines brand with 'dev' to signal developer tools focus
  ntstack.com (.com) — Abbreviated, professional, 'stack' suggests platform

8 Using a real LLM with tools¤

When ready to run against a real model, swap the FakeDriver for a PydanticAIDriver:

import os

from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver

driver = PydanticAIDriver(
    model_name="openai:gpt-4o-mini",
)

result = await execute(
    operation=operation,
    context_store=store,
    default_runtime=driver,
    block_store=block_store,
    tool_registry=tool_registry,
    prompt_engine=PromptEngine(block_store=block_store),
    context_overrides={
        "brand_name": "NovaTech",
        "industry": "developer tools",
    },
)
How tool discovery works at runtime

When the runtime executes an operation, it calls tool_registry.retrieve(agent.default_tool_scope) to get the list of tool schemas available to that agent. These schemas are converted into LLM-native function calling definitions so the model can decide when to invoke a tool.

The tool invocation itself happens through the LLM provider's native function-calling mechanism — CrewMaster feeds the tool results back into the conversation until the agent produces a final response.

Environment variables

The PydanticAIDriver reads model credentials from standard environment variables: - OPENAI_API_KEY — for OpenAI models - ANTHROPIC_API_KEY — for Claude models - GOOGLE_API_KEY — for Gemini models

Set these before running with a real driver.

9 Complete script¤

agent_with_tool.py 
"""Agent with tool — a domain name generator using ToolRegistry."""
import asyncio
import tempfile
from pathlib import Path

from pydantic import BaseModel, Field

from crewmaster import execute
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.engine import PromptEngine
from crewmaster.agents.prompts.local_disk_store import LocalDiskStore
from crewmaster.execution.runtime import (
    RuntimeRequest,
    RuntimeResponse,
    RuntimeStreamChunk,
)
from crewmaster.operations.operation import Operation
from crewmaster.tools.models import ToolSchema
from crewmaster.tools.registry import ToolRegistry


# 1. Output models
class DomainSuggestion(BaseModel):
    domain: str = Field(description="The suggested domain name")
    tld: str = Field(description="Top-level domain, e.g. .com, .io")
    rationale: str = Field(description="Why this domain fits the brand")


class DomainSuggestions(BaseModel):
    brand: str = Field(description="The brand name that was analyzed")
    suggestions: list[DomainSuggestion]


# 2. Tool schemas
WORD_LOOKUP_TOOL = ToolSchema(
    name="word_lookup",
    description="Look up related keywords and synonyms for a given term",
    input_schema={
        "type": "object",
        "properties": {
            "term": {"type": "string", "description": "Term to find related words for"},
            "max_results": {"type": "integer", "default": 5},
        },
        "required": ["term"],
    },
)

DOMAIN_CHECK_TOOL = ToolSchema(
    name="domain_check",
    description="Check availability and pricing for a domain name",
    input_schema={
        "type": "object",
        "properties": {
            "domain": {"type": "string", "description": "Domain name without TLD"},
            "tld": {"type": "string", "default": ".com"},
        },
        "required": ["domain"],
    },
)

# 3. Tool registry
tool_registry = ToolRegistry()
tool_registry.register("domain_naming", WORD_LOOKUP_TOOL)
tool_registry.register("domain_naming", DOMAIN_CHECK_TOOL)
tool_registry.register("*", WORD_LOOKUP_TOOL)
tool_registry.register("*", DOMAIN_CHECK_TOOL)

# 4. Agent
agent = AgentConfig(
    name="domain_namer",
    identity_blocks=["blocks://domain_namer/identity"],
    default_tool_scope="domain_naming",
)

# 5. Operation
operation = Operation(
    name="generate_domains",
    produces=DomainSuggestions,
    kind="artifact",
    agent=agent,
    task_blocks=["blocks://domain_namer/task"],
)

# 6. Prompt blocks
blocks_dir = Path(tempfile.mkdtemp())
(blocks_dir / "domain_namer" / "identity.j2").mkdir(parents=True)
(blocks_dir / "domain_namer" / "identity.j2").write_text("""\
---
id: blocks://domain_namer/identity
kind: identity
provides: domain_namer_persona
---
You are a creative brand naming expert. You use tools to research
keywords and check domain availability before making suggestions.
Always explain why each domain name fits the brand.
""")

(blocks_dir / "domain_namer" / "task.j2").write_text("""\
---
id: blocks://domain_namer/task
kind: task
provides: domain_suggestions_output
---
Generate 3 creative domain name suggestions for the brand "{{ brand_name }}"
in the {{ industry }} industry.

Steps:
1. Use word_lookup to find keywords related to "{{ brand_name }}" and "{{ industry }}"
2. Brainstorm domain names combining those keywords
3. Use domain_check to verify availability for your top picks
4. Return the suggestions as structured output
""")

block_store = LocalDiskStore(root=str(blocks_dir))

# 7. Context
store = ContextStore()


# 8. Fake driver
class FakeDriver:
    def __init__(self, output: DomainSuggestions) -> None:
        self._output = output
        self.calls: list[RuntimeRequest] = []

    async def execute(self, request: RuntimeRequest) -> RuntimeResponse:
        self.calls.append(request)
        return RuntimeResponse(output=self._output)

    async def astream(self, request: RuntimeRequest):
        self.calls.append(request)
        yield RuntimeStreamChunk(kind="text_delta", content="Researching...")
        yield RuntimeStreamChunk(kind="final", output=self._output)


# 9. Execute
async def main() -> None:
    driver = FakeDriver(
        output=DomainSuggestions(
            brand="NovaTech",
            suggestions=[
                DomainSuggestion(
                    domain="novatech.io",
                    tld=".io",
                    rationale="Short, tech-forward, matches the brand",
                ),
                DomainSuggestion(
                    domain="novadev.com",
                    tld=".com",
                    rationale="Combines brand with 'dev'",
                ),
                DomainSuggestion(
                    domain="ntstack.com",
                    tld=".com",
                    rationale="Abbreviated, professional",
                ),
            ],
        )
    )

    result = await execute(
        operation=operation,
        context_store=store,
        default_runtime=driver,
        block_store=block_store,
        tool_registry=tool_registry,
        prompt_engine=PromptEngine(block_store=block_store),
        context_overrides={
            "brand_name": "NovaTech",
            "industry": "developer tools",
        },
    )

    for s in result.suggestions:
        print(f"  {s.domain} ({s.tld}) — {s.rationale}")


asyncio.run(main())

Run it:

python agent_with_tool.py

  novatech.io (.io) — Short, tech-forward, matches the brand
  novadev.com (.com) — Combines brand with 'dev'
  ntstack.com (.com) — Abbreviated, professional

Next steps¤

  • Chain multiple tool-equipped operations in a DAG with the multi-node pipeline guide.
  • Learn about dynamic tool discovery with the request_capability mechanism.
  • Read the Tools API reference for full documentation on ToolSchema, Capability, and ToolRegistry.