Skip to content

naqsha.tools

Decorator-Driven API and tool execution

naqsha.tools

Tool contracts and starter tool implementations.

FunctionTool

Small adapter for simple Python callables.

Source code in src/naqsha/tools/base.py 
60
61
62
63
64
65
66
67
68
class FunctionTool:
    """Small adapter for simple Python callables."""

    def __init__(self, spec: ToolSpec, fn: Callable[[dict[str, Any]], ToolObservation]) -> None:
        self.spec = spec
        self._fn = fn

    def execute(self, arguments: dict[str, Any]) -> ToolObservation:
        return self._fn(arguments)

spec instance-attribute 

spec = spec

execute 

execute(arguments: dict[str, Any]) -> ToolObservation
Source code in src/naqsha/tools/base.py 
67
68
def execute(self, arguments: dict[str, Any]) -> ToolObservation:
    return self._fn(arguments)

Tool

Bases: Protocol

Source code in src/naqsha/tools/base.py 
53
54
55
56
57
class Tool(Protocol):
    spec: ToolSpec

    def execute(self, arguments: dict[str, Any]) -> ToolObservation:
        """Run the tool with validated arguments."""

spec instance-attribute 

spec: ToolSpec

execute 

execute(arguments: dict[str, Any]) -> ToolObservation

Run the tool with validated arguments.

Source code in src/naqsha/tools/base.py 
56
57
def execute(self, arguments: dict[str, Any]) -> ToolObservation:
    """Run the tool with validated arguments."""

ToolObservation dataclass

Source code in src/naqsha/tools/base.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@dataclass(frozen=True)
class ToolObservation:
    ok: bool
    content: str
    metadata: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        return {"ok": self.ok, "content": self.content, "metadata": self.metadata or {}}

    @classmethod
    def from_trace_payload(cls, payload: dict[str, Any]) -> ToolObservation:
        """Rebuild a persisted QAOA observation dict into a ToolObservation."""

        if not isinstance(payload, dict):
            raise ValueError("Observation payload must be an object.")
        ok = payload.get("ok")
        if not isinstance(ok, bool):
            raise ValueError("Observation 'ok' must be a boolean.")
        content = payload.get("content")
        if not isinstance(content, str):
            raise ValueError("Observation 'content' must be a string.")
        meta = payload.get("metadata")
        if meta is not None and not isinstance(meta, dict):
            raise ValueError("Observation 'metadata' must be an object when present.")
        return cls(ok=ok, content=content, metadata=meta)

ok instance-attribute 

ok: bool

content instance-attribute 

content: str

metadata class-attribute instance-attribute 

metadata: dict[str, Any] | None = None

to_dict 

to_dict() -> dict[str, Any]
Source code in src/naqsha/tools/base.py 
32
33
def to_dict(self) -> dict[str, Any]:
    return {"ok": self.ok, "content": self.content, "metadata": self.metadata or {}}

from_trace_payload classmethod 

from_trace_payload(
    payload: dict[str, Any],
) -> ToolObservation

Rebuild a persisted QAOA observation dict into a ToolObservation.

Source code in src/naqsha/tools/base.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def from_trace_payload(cls, payload: dict[str, Any]) -> ToolObservation:
    """Rebuild a persisted QAOA observation dict into a ToolObservation."""

    if not isinstance(payload, dict):
        raise ValueError("Observation payload must be an object.")
    ok = payload.get("ok")
    if not isinstance(ok, bool):
        raise ValueError("Observation 'ok' must be a boolean.")
    content = payload.get("content")
    if not isinstance(content, str):
        raise ValueError("Observation 'content' must be a string.")
    meta = payload.get("metadata")
    if meta is not None and not isinstance(meta, dict):
        raise ValueError("Observation 'metadata' must be an object when present.")
    return cls(ok=ok, content=content, metadata=meta)

ToolSpec dataclass

Source code in src/naqsha/tools/base.py 
17
18
19
20
21
22
23
@dataclass(frozen=True)
class ToolSpec:
    name: str
    description: str
    parameters: dict[str, Any]
    risk_tier: RiskTier = RiskTier.READ_ONLY
    read_only: bool = True

name instance-attribute 

name: str

description instance-attribute 

description: str

parameters instance-attribute 

parameters: dict[str, Any]

risk_tier class-attribute instance-attribute 

risk_tier: RiskTier = READ_ONLY

read_only class-attribute instance-attribute 

read_only: bool = True

AgentContext dataclass

Runtime context injected into tools that request it.

This is the stable public API surface for tool authors. Tools can access runtime state by including a ctx: AgentContext parameter in their signature. The Core Runtime will automatically inject this context at execution time.

Fields

shared_memory: Memory namespace accessible by all agents in a team (V2). For V2 Dynamic Memory Engine, this is a MemoryScope. For V1 compatibility, this may be a MemoryPort. private_memory: Memory namespace isolated to this specific agent (V2). For V2 Dynamic Memory Engine, this is a MemoryScope. For V1 compatibility, this may be a MemoryPort. span: Current trace span for attribution (V2). workspace_path: Absolute path to the Team Workspace root. agent_id: Unique identifier for the current agent. run_id: Unique identifier for the current run.

Source code in src/naqsha/tools/context.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True)
class AgentContext:
    """
    Runtime context injected into tools that request it.

    This is the stable public API surface for tool authors. Tools can access
    runtime state by including a `ctx: AgentContext` parameter in their signature.
    The Core Runtime will automatically inject this context at execution time.

    Fields:
        shared_memory: Memory namespace accessible by all agents in a team (V2).
                      For V2 Dynamic Memory Engine, this is a MemoryScope.
                      For V1 compatibility, this may be a MemoryPort.
        private_memory: Memory namespace isolated to this specific agent (V2).
                       For V2 Dynamic Memory Engine, this is a MemoryScope.
                       For V1 compatibility, this may be a MemoryPort.
        span: Current trace span for attribution (V2).
        workspace_path: Absolute path to the Team Workspace root.
        agent_id: Unique identifier for the current agent.
        run_id: Unique identifier for the current run.
    """

    shared_memory: MemoryScope | MemoryPort | None
    private_memory: MemoryScope | MemoryPort | None
    span: Span | None
    workspace_path: Path
    agent_id: str
    run_id: str

shared_memory instance-attribute 

shared_memory: MemoryScope | MemoryPort | None

private_memory instance-attribute 

private_memory: MemoryScope | MemoryPort | None

span instance-attribute 

span: Span | None

workspace_path instance-attribute 

workspace_path: Path

agent_id instance-attribute 

agent_id: str

run_id instance-attribute 

run_id: str

RiskTier

Bases: StrEnum

Tool risk classification for Tool Policy enforcement.

Source code in src/naqsha/tools/decorator.py 
15
16
17
18
19
20
class RiskTier(StrEnum):
    """Tool risk classification for Tool Policy enforcement."""

    READ_ONLY = "read_only"
    WRITE = "write"
    HIGH = "high"

READ_ONLY class-attribute instance-attribute 

READ_ONLY = 'read_only'

WRITE class-attribute instance-attribute 

WRITE = 'write'

HIGH class-attribute instance-attribute 

HIGH = 'high'

ToolDefinitionError

Bases: Exception

Raised at decoration time when a tool function signature is malformed.

Source code in src/naqsha/tools/decorator.py 
23
24
class ToolDefinitionError(Exception):
    """Raised at decoration time when a tool function signature is malformed."""

agent

Namespace for the @agent.tool decorator.

Source code in src/naqsha/tools/decorator.py 
223
224
225
226
class agent:
    """Namespace for the @agent.tool decorator."""

    tool = staticmethod(tool)

tool class-attribute instance-attribute 

tool = staticmethod(tool)

ToolExecutor

Executes decorated tools with automatic AgentContext injection.

The executor inspects signatures, injects AgentContext when requested via type hints, handles sync and async tool functions, and wraps exceptions as failed ToolObservation values so the runtime can enforce circuit breaker policy without crashing the agent loop.

Source code in src/naqsha/tools/executor.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class ToolExecutor:
    """
    Executes decorated tools with automatic AgentContext injection.

    The executor inspects signatures, injects ``AgentContext`` when requested via
    type hints, handles sync and async tool functions, and wraps exceptions as
    failed ``ToolObservation`` values so the runtime can enforce circuit breaker
    policy without crashing the agent loop.
    """

    def __init__(self, context: AgentContext) -> None:
        self._context = context

    def execute(
        self,
        func: Callable[..., Any],
        arguments: dict[str, Any],
    ) -> ToolObservation:
        """
        Execute a tool function with automatic context injection.

        For async tools, returns a coroutine that must be awaited.
        """

        sig = inspect.signature(func)
        try:
            type_hints = get_type_hints(func)
        except Exception:
            type_hints = {}
        needs_context = False
        for param_name, param in sig.parameters.items():
            ann = type_hints.get(param_name, param.annotation)
            if ann is AgentContext:
                needs_context = True
                break

        call_kwargs = arguments.copy()
        if needs_context:
            call_kwargs["ctx"] = self._context

        try:
            if inspect.iscoroutinefunction(func):
                return self._execute_async(func, call_kwargs)
            result = func(**call_kwargs)
        except Exception as exc:
            return tool_error_observation(exc)

        return self._convert_result(result)

    async def _execute_async(
        self,
        func: Callable[..., Any],
        call_kwargs: dict[str, Any],
    ) -> ToolObservation:
        """Execute an async tool and convert the result."""
        try:
            result = await func(**call_kwargs)
        except Exception as exc:
            return tool_error_observation(exc)

        return self._convert_result(result)

    def _convert_result(self, result: Any) -> ToolObservation:
        """Convert a tool result to ToolObservation."""
        if isinstance(result, ToolObservation):
            return result
        if isinstance(result, str):
            return ToolObservation(ok=True, content=result)
        if isinstance(result, dict):
            return ToolObservation(
                ok=result.get("ok", True),
                content=result.get("content", ""),
                metadata=result.get("metadata"),
            )
        return ToolObservation(ok=True, content=str(result))

execute 

execute(
    func: Callable[..., Any], arguments: dict[str, Any]
) -> ToolObservation

Execute a tool function with automatic context injection.

For async tools, returns a coroutine that must be awaited.

Source code in src/naqsha/tools/executor.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def execute(
    self,
    func: Callable[..., Any],
    arguments: dict[str, Any],
) -> ToolObservation:
    """
    Execute a tool function with automatic context injection.

    For async tools, returns a coroutine that must be awaited.
    """

    sig = inspect.signature(func)
    try:
        type_hints = get_type_hints(func)
    except Exception:
        type_hints = {}
    needs_context = False
    for param_name, param in sig.parameters.items():
        ann = type_hints.get(param_name, param.annotation)
        if ann is AgentContext:
            needs_context = True
            break

    call_kwargs = arguments.copy()
    if needs_context:
        call_kwargs["ctx"] = self._context

    try:
        if inspect.iscoroutinefunction(func):
            return self._execute_async(func, call_kwargs)
        result = func(**call_kwargs)
    except Exception as exc:
        return tool_error_observation(exc)

    return self._convert_result(result)

ToolRegistry

Registry for tools decorated with @agent.tool.

The registry: - Holds references to decorated tool functions - Supports lookup by name - Exports schema lists for Model Adapters - Validates that registered functions are properly decorated

Source code in src/naqsha/tools/registry.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class ToolRegistry:
    """
    Registry for tools decorated with @agent.tool.

    The registry:
    - Holds references to decorated tool functions
    - Supports lookup by name
    - Exports schema lists for Model Adapters
    - Validates that registered functions are properly decorated
    """

    def __init__(self) -> None:
        self._tools: dict[str, Callable[..., Any]] = {}

    def register(self, func: Callable[..., Any]) -> None:
        """
        Register a tool function.

        Args:
            func: A function decorated with @agent.tool

        Raises:
            ValueError: If the function is not decorated or name conflicts
        """
        if not hasattr(func, "__is_naqsha_tool__"):
            raise ValueError(
                f"Function {func.__name__} must be decorated with @agent.tool"
            )

        if not hasattr(func, "__tool_schema__"):
            raise ValueError(
                f"Function {func.__name__} is missing __tool_schema__ attribute"
            )

        schema = func.__tool_schema__
        name = schema["name"]

        if name in self._tools:
            raise ValueError(f"Tool '{name}' is already registered")

        self._tools[name] = func

    def get(self, name: str) -> Callable[..., Any] | None:
        """
        Look up a tool by name.

        Args:
            name: Tool name

        Returns:
            The tool function, or None if not found
        """
        return self._tools.get(name)

    def has(self, name: str) -> bool:
        """Check if a tool is registered."""
        return name in self._tools

    def names(self) -> frozenset[str]:
        """Return the set of registered tool names."""
        return frozenset(self._tools.keys())

    def export_schemas(self) -> list[dict[str, Any]]:
        """
        Export tool schemas for Model Adapters.

        Returns a list of schema dictionaries suitable for inclusion in
        model API requests. Each schema includes name, description, and
        parameters.

        Returns:
            List of tool schemas
        """
        schemas = []
        for func in self._tools.values():
            schema = func.__tool_schema__.copy()
            schemas.append(schema)
        return schemas

    def get_risk_tier(self, name: str) -> RiskTier | None:
        """
        Get the risk tier for a tool.

        Args:
            name: Tool name

        Returns:
            The tool's risk tier, or None if not found
        """
        func = self._tools.get(name)
        if func is None:
            return None
        return func.__tool_risk_tier__

    def is_read_only(self, name: str) -> bool:
        """
        Check if a tool is read-only.

        Args:
            name: Tool name

        Returns:
            True if the tool is read-only, False otherwise
        """
        func = self._tools.get(name)
        if func is None:
            return False
        return func.__tool_read_only__

    def clear(self) -> None:
        """Clear all registered tools (primarily for testing)."""
        self._tools.clear()

register 

register(func: Callable[..., Any]) -> None

Register a tool function.

Parameters:

Name Type Description Default
func Callable[..., Any]

A function decorated with @agent.tool

 required

Raises:

Type Description
ValueError

If the function is not decorated or name conflicts
Source code in src/naqsha/tools/registry.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def register(self, func: Callable[..., Any]) -> None:
    """
    Register a tool function.

    Args:
        func: A function decorated with @agent.tool

    Raises:
        ValueError: If the function is not decorated or name conflicts
    """
    if not hasattr(func, "__is_naqsha_tool__"):
        raise ValueError(
            f"Function {func.__name__} must be decorated with @agent.tool"
        )

    if not hasattr(func, "__tool_schema__"):
        raise ValueError(
            f"Function {func.__name__} is missing __tool_schema__ attribute"
        )

    schema = func.__tool_schema__
    name = schema["name"]

    if name in self._tools:
        raise ValueError(f"Tool '{name}' is already registered")

    self._tools[name] = func

get 

get(name: str) -> Callable[..., Any] | None

Look up a tool by name.

Parameters:

Name Type Description Default
name str

Tool name

 required

Returns:

Type Description
Callable[..., Any] | None

The tool function, or None if not found
Source code in src/naqsha/tools/registry.py 
53
54
55
56
57
58
59
60
61
62
63
def get(self, name: str) -> Callable[..., Any] | None:
    """
    Look up a tool by name.

    Args:
        name: Tool name

    Returns:
        The tool function, or None if not found
    """
    return self._tools.get(name)

has 

has(name: str) -> bool

Check if a tool is registered.

Source code in src/naqsha/tools/registry.py 
65
66
67
def has(self, name: str) -> bool:
    """Check if a tool is registered."""
    return name in self._tools

names 

names() -> frozenset[str]

Return the set of registered tool names.

Source code in src/naqsha/tools/registry.py 
69
70
71
def names(self) -> frozenset[str]:
    """Return the set of registered tool names."""
    return frozenset(self._tools.keys())

export_schemas 

export_schemas() -> list[dict[str, Any]]

Export tool schemas for Model Adapters.

Returns a list of schema dictionaries suitable for inclusion in model API requests. Each schema includes name, description, and parameters.

Returns:

Type Description
list[dict[str, Any]]

List of tool schemas
Source code in src/naqsha/tools/registry.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def export_schemas(self) -> list[dict[str, Any]]:
    """
    Export tool schemas for Model Adapters.

    Returns a list of schema dictionaries suitable for inclusion in
    model API requests. Each schema includes name, description, and
    parameters.

    Returns:
        List of tool schemas
    """
    schemas = []
    for func in self._tools.values():
        schema = func.__tool_schema__.copy()
        schemas.append(schema)
    return schemas

get_risk_tier 

get_risk_tier(name: str) -> RiskTier | None

Get the risk tier for a tool.

Parameters:

Name Type Description Default
name str

Tool name

 required

Returns:

Type Description
RiskTier | None

The tool's risk tier, or None if not found
Source code in src/naqsha/tools/registry.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def get_risk_tier(self, name: str) -> RiskTier | None:
    """
    Get the risk tier for a tool.

    Args:
        name: Tool name

    Returns:
        The tool's risk tier, or None if not found
    """
    func = self._tools.get(name)
    if func is None:
        return None
    return func.__tool_risk_tier__

is_read_only 

is_read_only(name: str) -> bool

Check if a tool is read-only.

Parameters:

Name Type Description Default
name str

Tool name

 required

Returns:

Type Description
bool

True if the tool is read-only, False otherwise
Source code in src/naqsha/tools/registry.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def is_read_only(self, name: str) -> bool:
    """
    Check if a tool is read-only.

    Args:
        name: Tool name

    Returns:
        True if the tool is read-only, False otherwise
    """
    func = self._tools.get(name)
    if func is None:
        return False
    return func.__tool_read_only__

clear 

clear() -> None

Clear all registered tools (primarily for testing).

Source code in src/naqsha/tools/registry.py 
120
121
122
def clear(self) -> None:
    """Clear all registered tools (primarily for testing)."""
    self._tools.clear()

validate_arguments 

validate_arguments(
    schema: dict[str, Any], arguments: dict[str, Any]
) -> None

Validate a conservative subset of JSON Schema used by starter tools.

Source code in src/naqsha/tools/base.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def validate_arguments(schema: dict[str, Any], arguments: dict[str, Any]) -> None:
    """Validate a conservative subset of JSON Schema used by starter tools."""

    if schema.get("type") != "object":
        raise ValueError("Tool parameter schema must be an object schema.")
    properties = schema.get("properties", {})
    required = set(schema.get("required", []))
    missing = required - set(arguments)
    if missing:
        raise ValueError(f"Missing required tool arguments: {sorted(missing)}")
    unexpected = set(arguments) - set(properties)
    if schema.get("additionalProperties") is False and unexpected:
        raise ValueError(f"Unexpected tool arguments: {sorted(unexpected)}")
    for name, value in arguments.items():
        prop = properties.get(name, {})
        expected = prop.get("type")
        if expected == "string" and not isinstance(value, str):
            raise ValueError(f"Tool argument '{name}' must be a string.")
        if expected == "integer":
            if not isinstance(value, int) or isinstance(value, bool):
                raise ValueError(f"Tool argument '{name}' must be an integer.")
        if expected == "number":
            if isinstance(value, bool) or not isinstance(value, int | float):
                raise ValueError(f"Tool argument '{name}' must be a number.")
        if expected == "boolean" and not isinstance(value, bool):
            raise ValueError(f"Tool argument '{name}' must be a boolean.")
        if expected == "array":
            if not isinstance(value, list):
                raise ValueError(f"Tool argument '{name}' must be an array.")
            min_items = prop.get("minItems")
            if isinstance(min_items, int) and len(value) < min_items:
                raise ValueError(
                    f"Tool argument '{name}' must contain at least {min_items} elements."
                )
            items_schema = prop.get("items", {})
            item_type = items_schema.get("type")
            for i, item in enumerate(value):
                if item_type == "string" and not isinstance(item, str):
                    raise ValueError(f"Tool argument '{name}[{i}]' must be a string.")

decorated_to_function_tool 

decorated_to_function_tool(
    func: Callable[..., Any],
    get_context: Callable[[], AgentContext],
) -> FunctionTool

Wrap a decorated tool function as a FunctionTool with runtime context injection.

Source code in src/naqsha/tools/decorated_adapter.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def decorated_to_function_tool(
    func: Callable[..., Any],
    get_context: Callable[[], AgentContext],
) -> FunctionTool:
    """Wrap a decorated tool function as a ``FunctionTool`` with runtime context injection."""
    schema = func.__tool_schema__
    dec_tier = func.__tool_risk_tier__
    tier = _coerce_policy_risk_tier(dec_tier)
    spec = ToolSpec(
        name=schema["name"],
        description=schema["description"],
        parameters=schema["parameters"],
        risk_tier=tier,
        read_only=tier == RiskTier.READ_ONLY,
    )

    def execute(arguments: dict[str, Any]) -> ToolObservation:
        executor = ToolExecutor(get_context())
        result = executor.execute(func, arguments)
        if inspect.iscoroutine(result):
            return asyncio.run(result)
        return cast("ToolObservation", result)

    return FunctionTool(spec, execute)

tool 

tool(
    *,
    risk_tier: RiskTier = RiskTier.READ_ONLY,
    description: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator for defining tools using the Decorator-Driven API.

Usage

@agent.tool(risk_tier=RiskTier.WRITE, description="Write a file") def write_file(path: str, content: str, ctx: AgentContext) -> str: '''Write content to a file.''' # Implementation return "File written"

The decorator: 1. Generates a JSON Schema from the function's type hints and docstring 2. Stores the schema on the function as __tool_schema__ 3. Marks the function as a tool with __is_naqsha_tool__ = True 4. Stores the risk tier as __tool_risk_tier__ 5. Stores whether the tool is read-only as __tool_read_only__

Parameters with type hint AgentContext are automatically injected at runtime and omitted from the public schema.

Parameters:

Name Type Description Default
risk_tier RiskTier

Tool risk classification (READ_ONLY, WRITE, HIGH)

 READ_ONLY
description str | None

Optional description (defaults to first line of docstring)

 None

Raises:

Type Description
ToolDefinitionError

If the function signature is malformed
Source code in src/naqsha/tools/decorator.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def tool(
    *,
    risk_tier: RiskTier = RiskTier.READ_ONLY,
    description: str | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """
    Decorator for defining tools using the Decorator-Driven API.

    Usage:
        @agent.tool(risk_tier=RiskTier.WRITE, description="Write a file")
        def write_file(path: str, content: str, ctx: AgentContext) -> str:
            '''Write content to a file.'''
            # Implementation
            return "File written"

    The decorator:
    1. Generates a JSON Schema from the function's type hints and docstring
    2. Stores the schema on the function as `__tool_schema__`
    3. Marks the function as a tool with `__is_naqsha_tool__ = True`
    4. Stores the risk tier as `__tool_risk_tier__`
    5. Stores whether the tool is read-only as `__tool_read_only__`

    Parameters with type hint `AgentContext` are automatically injected at runtime
    and omitted from the public schema.

    Args:
        risk_tier: Tool risk classification (READ_ONLY, WRITE, HIGH)
        description: Optional description (defaults to first line of docstring)

    Raises:
        ToolDefinitionError: If the function signature is malformed
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        if not callable(func):
            raise ToolDefinitionError(f"{func} is not callable")

        try:
            schema = _generate_schema(func, description)
        except ToolDefinitionError:
            raise
        except Exception as exc:
            raise ToolDefinitionError(
                f"Failed to generate schema for {func.__name__}: {exc}"
            ) from exc

        func.__tool_schema__ = schema  # type: ignore[attr-defined]
        func.__is_naqsha_tool__ = True  # type: ignore[attr-defined]
        func.__tool_risk_tier__ = risk_tier  # type: ignore[attr-defined]
        func.__tool_read_only__ = risk_tier == RiskTier.READ_ONLY  # type: ignore[attr-defined]

        return func

    return decorator