How to Build and Register Custom Tools for Agent Execution in OpenRisk

To build and register custom tools for agent execution in OpenRisk, define a Python function with type hints, decorate it with @system_tool from derisk.agent.core.system_tool_registry, and ensure the module is imported so the agent's system_tool_injection() method can expose it to the LLM.

OpenRisk provides a registry-based plug-in system that allows developers to extend agent capabilities without modifying core framework code. By registering custom tools, you enable LLM agents to invoke domain-specific logic, external APIs, or data retrieval operations during task execution.

Understanding the Tool Registration Architecture

OpenRisk’s tool system relies on three core components: a global registry, a function wrapper, and agent-level injection. Understanding these components ensures you implement tools that integrate seamlessly with the agent execution loop.

The System Tool Registry

The system_tool_registry.py file in packages/derisk-core/src/derisk/agent/core/ houses the global system_tool_dict and implements the @system_tool decorator. When you decorate a function, the decorator creates a FunctionTool instance and stores it in system_tool_dict under the specified tool name. This registry acts as the single source of truth for all available tools before they are assigned to specific agents.

The FunctionTool Wrapper

FunctionTool (imported from derisk.agent.resource.tool.base) wraps your original Python callable, providing standardized entry points for synchronous (execute) and asynchronous (async_execute) invocation. It also handles JSON schema generation from your function’s type annotations, manages streaming output if enabled, and enforces concurrency policies. This abstraction ensures that regardless of how you write your tool function, the agent interacts with it through a consistent interface.

Agent Integration via BaseAgent

The BaseAgent class in base_agent.py initializes each agent instance by calling system_tool_injection(). This method copies all entries from the global system_tool_dict into the instance’s available_system_tools mapping. When an agent processes a function_call request from the LLM, it resolves the tool name against self.available_system_tools and invokes the wrapper’s execution method. Concrete implementations like react_agent.py demonstrate this injection pattern in practice.

Step-by-Step Guide to Building Custom Tools

Follow these steps to create, register, and verify a custom tool within the OpenRisk framework.

Step 1: Create the Tool Module

Create a new Python file in packages/derisk-core/src/derisk/agent/core/tools/ or any importable location within your project. Name the file descriptively, such as reverse_tool.py or data_fetcher.py.

Step 2: Define the Function with Type Annotations

Write your tool as a standard Python function. Use type hints for all parameters and the return value. For human-readable documentation that appears in the LLM prompt, use typing.Annotated to attach descriptions to parameters.

from typing import Annotated, List

def summarize_text(
    content: Annotated[str, "The full markdown content to summarize"],
    max_sentences: Annotated[int, "Number of bullet points to return"] = 5,
) -> List[str]:
    """Simple heuristic summarizer."""
    # Implementation logic here

    return []

Step 3: Apply the @system_tool Decorator

Import the decorator from the registry and apply it to your function. Configure execution behavior using the decorator’s parameters:

from derisk.agent.core.system_tool_registry import system_tool

@system_tool(
    name="summarize_text",
    description="Summarizes a long markdown string into a concise bullet list.",
    concurrency="parallel",  # or "sequential"

    stream=False,            # set True for streaming generators

    ask_user=False,          # True if tool requires operator confirmation

)
def summarize_text(
    content: Annotated[str, "The full markdown content to summarize"],
    max_sentences: Annotated[int, "Number of bullet points to return"] = 5,
) -> List[str]:
    """Simple heuristic summarizer."""
    return content.split(". ")[:max_sentences]

The decorator automatically registers the tool in system_tool_dict upon module import.

Step 4: Verify Registration

Ensure your tool module is imported during application startup. Then verify registration programmatically:

from derisk.agent.core.system_tool_registry import system_tool_dict

assert "summarize_text" in system_tool_dict
print(system_tool_dict["summarize_text"].execute("First sentence. Second sentence.", max_sentences=2))

Complete Working Example

Below is a minimal, self-contained custom tool that reverses a string. This demonstrates the full implementation pattern including type annotations, decorator usage, and error handling:

# packages/derisk-core/src/derisk/agent/core/tools/reverse_tool.py

from typing import Annotated
from derisk.agent.core.system_tool_registry import system_tool

@system_tool(
    name="reverse_string",
    description="Reverses the input text. Useful for simple obfuscation or palindrome checks.",
    concurrency="sequential",
    stream=False,
)
def reverse_string(
    text: Annotated[str, "The original string to reverse"]
) -> str:
    """Return the reversed version of text."""
    return text[::-1]

After importing this module, agents can invoke the tool:

from derisk.agent.core.system_tool_registry import system_tool_dict

tool = system_tool_dict["reverse_string"]
result = tool.execute("hello")  # Returns "olleh"

Advanced Configuration Options

Fine-tune tool behavior using decorator parameters to match execution requirements.

Concurrency Control

Set concurrency="parallel" to allow simultaneous executions, or concurrency="sequential" to enforce one-at-a-time access. Sequential mode prevents race conditions when tools modify shared state or access rate-limited external APIs.

Streaming Execution

Enable stream=True when your tool produces partial results over time (e.g., reading a large file line-by-line). The FunctionTool wrapper will route calls to execute_stream, yielding chunks back to the agent for progressive display.

User Confirmation Prompts

Set ask_user=True for destructive or sensitive operations. This flag signals the agent runtime to prompt the operator for confirmation before invoking the tool, adding a safety layer to custom tool execution.

Summary

• Use the @system_tool decorator from derisk.agent.core.system_tool_registry to register any Python callable as an agent tool.
• Include type annotations and Annotated parameter descriptions to auto-generate JSON schemas for the LLM.
• Place tool modules in packages/derisk-core/src/derisk/agent/core/tools/ or ensure they are imported at runtime to populate system_tool_dict.
• Agents automatically discover registered tools via BaseAgent.system_tool_injection(), which copies entries into available_system_tools for runtime resolution.
• Configure execution behavior using decorator parameters: concurrency for parallel/sequential control, stream for generator-based output, and ask_user for confirmation prompts.

Frequently Asked Questions

How does the agent discover tools registered with @system_tool?

The BaseAgent class calls system_tool_injection() during initialization. This method imports the global system_tool_dict from system_tool_registry.py and copies all registered FunctionTool instances into the agent’s self.available_system_tools dictionary. When the LLM requests a function call, the agent resolves the tool name against this dictionary to locate the executable wrapper.

Can I use async functions as custom tools?

Yes. The FunctionTool wrapper in derisk.agent.resource.tool.base provides both execute (synchronous) and async_execute (asynchronous) entry points. If your tool function is defined with async def, the wrapper automatically routes calls through async_execute. The agent runtime handles the event loop integration, allowing non-blocking I/O operations like HTTP requests or database queries within custom tools.

What is the difference between parallel and sequential concurrency?

The concurrency parameter in the @system_tool decorator controls execution isolation. When set to "parallel", multiple invocations of the tool can run simultaneously, suitable for stateless computations or read-only operations. When set to "sequential", the FunctionTool enforces a lock or queue mechanism to ensure only one execution runs at a time, preventing race conditions when the tool modifies shared resources or accesses rate-limited external services.

How do I verify that my custom tool is properly registered?

Import the system_tool_dict from derisk.agent.core.system_tool_registry and check for your tool’s name as a key. You can also inspect the FunctionTool instance to verify its schema and execution method. For runtime verification, initialize an agent that calls system_tool_injection() and inspect agent.available_system_tools to confirm the tool appears in the agent’s local registry before the LLM attempts to invoke it.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}