Tool Interface Reference

Decorator Tools

The simplest way to create a tool:

@beam_tool(name="my_tool", timeout=300)
async def my_tool(input_paths: list, output_path: Path, parameters: dict) -> dict:
    target = parameters.get("target")
    return {"result": "data"}

Parameter	Type	Description
input_paths	list	Input files from previous workflow nodes. Empty list if none.
output_path	Path	Where to write results.
parameters	dict	Configuration from the workflow node.

Return a dict → automatically saved as JSON output. Return None → you handle writing to output_path yourself.

---

Subclass Tools

For tools that need setup, cleanup, or state.

ExecutionContext

What your tool receives:

Field	Type	Description
input_paths	List[Path]	Input files (can be empty)
output_path	Path	Where to write results
parameters	dict	Workflow node configuration
execution_id	str	Unique execution ID
timeout_seconds	int	Execution timeout

ExecutionResult

What your tool returns:

Field	Required	Description
status	Yes	"success", "failure", or "timeout"
exit_code	Yes	0 for success
output_data	No	Output as dict
output_path	No	Path to output file
error	No	Error message on failure

Example

from beam_agent import BeamTool
from beam_agent.core.base import ExecutionContext, ExecutionResult

class MyTool(BeamTool):
    def __init__(self):
        super().__init__(name="my_tool", timeout=300)

    async def execute(self, context: ExecutionContext) -> ExecutionResult:
        # Your logic here
        with open(context.output_path, 'w') as f:
            json.dump({"result": "data"}, f)

        return ExecutionResult(status="success", exit_code=0)

Override setup() and cleanup() if you need to initialize/tear down resources (DB connections, HTTP sessions, etc).

---

Script Tools

Wrap existing scripts or executables without modifying them.

Factory Functions

bash_script("name", "/path/to/script.sh")
python_script("name", "/path/to/script.py")
powershell_script("name", "C:/Scripts/tool.ps1")
executable("name", "/opt/bin/scanner")

Placeholders

Use these in args_template to pass data to your script:

Placeholder	Value
{input}	First input file path
{output}	Output file path
{param.name}	Value from parameters["name"]
{env.NAME}	Environment variable $NAME

Custom Arguments

from beam_agent import BeamScriptTool, ScriptConfig

tool = BeamScriptTool(
    name="scanner",
    config=ScriptConfig(
        command=["python3", "/opt/scanner.py"],
        args_template=["--target", "{param.target}", "--out", "{output}"],
    ),
)

---

Built-in: IntegrationProxyTool

HTTP proxy for calling internal services from NINA workflows. Encryption is auto-configured.

from beam_agent.tools import IntegrationProxyTool
agent.register_tool("integration_proxy", IntegrationProxyTool())

By default returns only the API response body (body_only=True). Set body_only=False for full response metadata.

---

Next Steps

• Creating Tools - Step-by-step guide
• Example Tools - Working examples