sam-tool-sdk 0.1.0

Standalone SDK for developing Solace Agent Mesh Python tools

SAM Tool SDK

Standalone Python SDK for developing tools that run in Solace Agent Mesh (SAM).

What This Is

This package provides the minimal Python runtime and base classes needed to develop and execute Python tools for SAM. It is a self-contained extraction of the tool execution path from the full solace_agent_mesh Python package — just the pieces that matter for writing and running tools.

The Go SAM Secure Tool Runtime (STR) executes each tool as a subprocess by invoking the console script the tool's pyproject.toml registers (built on tool_cli from this SDK). This package is everything that subprocess needs.

Why It Exists

The Go implementation of SAM (solace-agent-mesh-go) originally required the full solace_agent_mesh Python package to be installed just to execute Python tools. That package pulls in dozens of dependencies (google-adk, solace-pubsubplus, onnxruntime, pydub, etc.) that are irrelevant to tool execution in a sandbox.

By extracting the ~1,500 lines of Python actually needed into this standalone SDK, we:

• Break the dependency — no need to install solace_agent_mesh or any of its heavy transitive dependencies
• Simplify installation — pip install sam-tool-sdk is all you need (only requires pydantic)
• Own the contract — the Go STR and this SDK share a well-defined interface (JSON files + named pipes) without coupling to the Python framework's internals
• Speed up sandbox startup — avoids importing the full agent framework, which takes ~5 seconds in a cold sandbox

What's Included

Module	Purpose
cli.py	tool_cli, dynamic_tool_cli, provider_cli — wrap a function or class as a console-script entry point with --schema + sandbox execution support
dynamic_tool.py	DynamicTool and DynamicToolProvider base classes for tool authors
tool_result.py	ToolResult, DataObject, DataDisposition — structured result types returned to the agent
artifact_types.py	Artifact dataclass + ArtifactTypeInfo helpers for declaring artifact parameters and pre-loading their content
context_facade.py	SandboxToolContextFacade — task_id, session_id, artifact APIs, status updates, and LLM callbacks available to tools at runtime
tool_options.py	tool_timeout, with_dynamic_schema, with_volume_params, VolumeParam, VolumeMount decorators for per-tool sandbox options
config_schema.py	ConfigSchemaField, with_config_schema — declare operator-supplied config (API keys, endpoints, defaults) rendered by the platform UI
ipc.py	IPCClient, IPCError — JSON-RPC client for the sandbox-to-host LLM callback channel
tool_runner.py	Sandbox entry point invoked by the Go STR; ordinarily reached via the tool_cli-built console script rather than directly

Installation

pip install sam-tool-sdk

The only runtime dependency is pydantic>=2.4.

For local development against the source checkout:

pip install -e .

(run from the python/sam-tool-sdk/ directory of solace-agent-mesh-go)

Quick Start

Function-based tool with tool_cli

The canonical pattern. Author a function, wrap it with tool_cli, expose the resulting callable as a console script via pyproject.toml. STR invokes the console script directly — --schema for tool discovery, with a runner_args.json path for execution.

# greet_tool.py
from sam_tool_sdk import tool_cli, ToolResult, SandboxToolContextFacade


async def greet(name: str, excited: bool = False, ctx: SandboxToolContextFacade = None) -> ToolResult:
    """Greet someone by name.

    Args:
        name: who to greet
        excited: add an exclamation mark
    """
    if ctx is not None:
        ctx.send_status(f"Greeting {name}...")
    suffix = "!" if excited else "."
    return ToolResult.ok(message=f"Hello, {name}{suffix}", data={"greeted": name})


cli = tool_cli(greet)

# pyproject.toml
[project]
name = "sam-tool-greet"
version = "0.1.0"
dependencies = ["sam-tool-sdk>=0.1,<0.2"]

[project.scripts]
greet = "greet_tool:cli"

The framework detects parameter type annotations (str, bool, int, Artifact, SandboxToolContextFacade, …), parses the docstring's Args: block for per-parameter descriptions, and emits a JSON Schema the LLM sees via --schema. The SandboxToolContextFacade-annotated parameter is injected at runtime; everything else becomes a tool argument.

Class-based tool (DynamicTool)

For more control over the tool's schema and behavior:

from sam_tool_sdk import DynamicTool, ToolResult

class WordCounter(DynamicTool):
    @property
    def tool_name(self):
        return "count_words"

    @property
    def tool_description(self):
        return "Counts the number of words in the given text."

    @property
    def parameters_schema(self):
        return {
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "The text to count words in"},
            },
            "required": ["text"],
        }

    async def _run_async_impl(self, args, tool_context=None, credential=None):
        count = len(args["text"].split())
        return ToolResult.ok(f"Found {count} words", data={"count": count}).model_dump()

Provider-based tools (DynamicToolProvider)

When you want to define multiple related tools from a single class:

from sam_tool_sdk import DynamicToolProvider

class MathTools(DynamicToolProvider):
    @MathTools.register_tool
    async def add(self, a: float, b: float) -> dict:
        """Add two numbers."""
        return {"text": str(a + b)}

    @MathTools.register_tool
    async def multiply(self, a: float, b: float) -> dict:
        """Multiply two numbers."""
        return {"text": str(a * b)}

    def create_tools(self, tool_config=None):
        return []  # All tools defined via @register_tool

Declaring operator config (config_schema)

Tools that need operator-supplied configuration (API keys, endpoints, default channels, …) declare it with @with_config_schema so the platform can render a config form, validate per-agent overrides, and enforce required fields at deploy time. The wire format matches the Go SDK's ConfigSchemaField exactly — --schema output is interchangeable between the two SDKs.

from sam_tool_sdk import ConfigSchemaField, with_config_schema

@with_config_schema([
    ConfigSchemaField(
        key="slack_bot_token",
        type="string",
        description="Slack Bot OAuth token (xoxb-...).",
        required=True,
        secret=True,
    ),
    ConfigSchemaField(key="default_channel", type="string"),
    ConfigSchemaField(key="api_url", type="string"),
])
async def post_message(tool_config, channel: str, text: str):
    token = tool_config["slack_bot_token"]
    ...

Values are resolved at deploy time with precedence: agent override > toolset default > schema default > 422 if a required field is unset. Fields marked secret=True are redacted on API responses and rendered with a password input in the UI.

For DynamicTool subclasses, override the config_schema property to return a list of ConfigSchemaField(...).to_dict() entries.

Working with artifacts

Tools can receive and produce artifacts using type annotations:

from sam_tool_sdk import Artifact, ToolResult, DataObject, DataDisposition

async def summarize_file(doc: Artifact) -> ToolResult:
    """Summarize the contents of an uploaded document."""
    text = doc.as_text()
    summary = text[:200] + "..."  # placeholder for real summarization

    return ToolResult.ok(
        message=f"Summarized {doc.filename}",
        data={"summary": summary},
        data_objects=[
            DataObject(
                name="summary.txt",
                content=summary,
                mime_type="text/plain",
                disposition=DataDisposition.ARTIFACT,
                description="Document summary",
            )
        ],
    )

Per-tool sandbox options

Decorate a tool_cli function (or override the matching DynamicTool properties) to declare timeouts and volume requirements that STR honors at execution time:

from sam_tool_sdk import tool_cli, tool_timeout, with_volume_params, VolumeParam, ToolResult


@tool_timeout(seconds=120)
@with_volume_params([
    VolumeParam(name="workspace", mount_path="/workspace", mode="readwrite"),
])
async def transform(input_path: str, ctx=None) -> ToolResult:
    """Long-running transformation that needs scratch space."""
    ...

cli = tool_cli(transform)

tool_timeout overrides the per-invocation timeout; with_volume_params declares mount points STR provisions before the tool runs (paths surface via ctx.volume_mounts / ctx.get_volume_mount_path("/workspace"), keyed by mount path).

Packaging

A SAM tool ships as an AWS-Lambda-Layer-style directory: python/bin/<console-script> plus all dependencies pre-extracted under python/. STR sets PYTHONPATH=<toolDir>/python:<toolDir> when launching the console script, so the interpreter finds sam_tool_sdk (and everything else) without a virtualenv.

Build that layout with a single pip install --target invocation:

pip install --target dist/python \
    --platform manylinux2014_x86_64 \
    --python-version 3.13 \
    --only-binary=:all: \
    .

The directory then contains dist/python/bin/<your-script>, dist/python/sam_tool_sdk/, dist/python/pydantic/, and your own package. Pair it with a manifest.yaml declaring executable: python/bin/<your-script> and STR will discover, schema-probe, and invoke the tool.

The three production Python tools under tools/python/ (plotly, pdf_to_markdown, python_executor) are working references for the full layout.

Relationship to solace_agent_mesh

This SDK is a standalone extraction of the tool execution path from the solace_agent_mesh Python package. Tools written with sam_tool_sdk are wire-compatible with both the Go and Python SAM runtimes — the JSON-based communication protocol (runner_args.json, result.json, status pipe) is identical.

Key difference: parameters_schema

In solace_agent_mesh, DynamicTool.parameters_schema returns a google.genai.types.Schema object. In sam_tool_sdk, it returns a standard JSON Schema dict:

# solace_agent_mesh (old)
from google.genai import types as adk_types
@property
def parameters_schema(self):
    return adk_types.Schema(
        type=adk_types.Type.OBJECT,
        properties={"name": adk_types.Schema(type=adk_types.Type.STRING)},
        required=["name"],
    )

# sam_tool_sdk (new)
@property
def parameters_schema(self):
    return {
        "type": "object",
        "properties": {"name": {"type": "string"}},
        "required": ["name"],
    }

Migrating from solace_agent_mesh

1. Replace imports: from solace_agent_mesh.agent.tools.dynamic_tool import DynamicTool becomes from sam_tool_sdk import DynamicTool
2. Change parameters_schema to return a JSON Schema dict instead of adk_types.Schema
3. Replace ToolContextFacade annotations with SandboxToolContextFacade (or just use the string name — both are detected)

How It Works

At discovery time, STR runs <your-console-script> --schema to learn the tool's name, description, parameter JSON Schema, declared timeout, and volume requirements. At invocation time, STR runs <your-console-script> <path-to-runner_args.json> — the console script (built by tool_cli) reads the args, imports your tool function or class, injects SandboxToolContextFacade plus any pre-loaded Artifact parameters, awaits the result, and writes result.json for STR to read back.

Full sequence for a single invocation:

1. STR receives an A2A sam_remote_tool/invoke request over the broker.
2. STR pre-loads any Artifact-typed parameters into a work directory.
3. STR writes runner_args.json (arguments, artifact metadata, status pipe path, IPC socket path, output paths, …).
4. STR spawns <your-console-script> /path/to/runner_args.json with PYTHONPATH set to include the toolset's python/ directory.
5. The console script (this SDK) imports the tool, detects type annotations, injects context + artifacts, and calls the tool.
6. The tool writes status updates to the named pipe (ctx.send_status(...)) and optionally calls back to the host for LLM completions over the IPC socket (ctx.call_llm(...)).
7. The tool returns a ToolResult or a dict; the SDK serialises any DataObject outputs to the artifact directory and writes result.json.
8. STR reads result.json and the artifact directory, then publishes the A2A sam_remote_tool/response.

License

Apache-2.0