Toolplane Python Client

The Python package is the primary maintained SDK for Toolplane's durable remote tool-execution control plane. Python is the richest current client surface and the baseline for the repo's end-to-end provider, consumer, and admin capability.

Decision Rule

Use Toolplane when one remote tool may outlive the caller, needs inspection or bounded replay after disconnect, needs explicit provider ownership or drain behavior, or is queue-backed enough that request lifecycle control matters. If the work is quick, in-process, same-lifecycle, and not operationally sensitive, direct tool calling is simpler.

A concrete first offload candidate is one sandboxed code-execution worker. The Python provider-consumer path is the maintained starting point for that shape because it covers explicit provider ownership, request lifecycle control, and the maintained HTTP gateway compatibility layer on the richest SDK surface.

Support Status

• Primary maintained SDK and current completeness baseline.
• gRPC is the primary transport for the repo's control-plane story.
• ToolplaneHTTP is a maintained compatibility surface over the same session, tool, machine, request, and task flows.
• Python exposes a broader public surface than the current Go and TypeScript SDKs; confirm cross-SDK portability in SDK_MAP.md before assuming parity.

First Offload Path

Start with the maintained provider-consumer pair before exploring the full API. The sample tools are intentionally simple, but this is the same lifecycle you would reuse to offload one sandboxed code-execution tool or another environment-bound worker:

1. Run example_client.py — connects via gRPC, creates a session through the explicit ProviderRuntime, registers machine-backed tools, and starts the provider loop.
2. Copy the printed TOOLPLANE_SESSION_ID.
3. Run example_user.py with that session ID — lists tools, invokes provider-backed work, and polls request state.

See README_EXAMPLES.md for environment defaults and the full example flow. For runtime semantics — request lifecycle, streaming, recovery windows, machine drain — see server/DOCUMENTATION.md.

Scope Categories

The Python public surface spans three scope categories (see SDK_MAP.md for the full matrix):

• Consumer scope — session/machine/task lifecycle, tool discovery, and remote invocation (create_session, invoke, stream, etc.). These methods are portable across maintained SDKs.
• Provider scope — tool registration, request claiming, heartbeat, and result submission through the explicit ProviderRuntime surface. The client also exposes convenience aliases tool(), start(), and stop() over that same runtime surface.
• Admin scope — session administration helpers (list_user_sessions, bulk_delete_sessions, get_session_stats, invalidate_session). These are currently exposed only in the Python SDK.

Installation

The package is not published to PyPI yet; install it from the repository.

# Clone the repository
git clone https://github.com/Abhishek-chohan/tool-control-plane.git
cd tool-control-plane/clients/python-client

# Install in development mode
pip install -e .

# Install with all development dependencies
pip install -e ".[dev]"

Requirements: Python 3.8+ with grpcio, protobuf, requests, googleapis-common-protos, and pydantic. The optional toolkits under toolplane/toolkits/ have heavier dependencies; install their requirements only if you use them. grpcio-tools is needed only for proto regeneration (cd server && make gen-proto-python).

Quick Start

Consumer Usage with gRPC Client

from toolplane import Toolplane

# Initialize a consumer-oriented client
client = Toolplane(
    server_host="localhost",
    server_port=9001,
    api_key="your-api-key-here",
)

client.connect()

# Attach to an existing provider-backed session
session_id = "provider-session-123"
tools = client.get_available_tools(session_id)
print(tools)

# Execute a tool synchronously
result = client.invoke("echo-tool", session_id, text="Hello World")
print(result)  # Output: Echo: Hello World

# Submit a tool invocation without blocking; awaits the request ID
import asyncio
request_id = asyncio.run(
    client.ainvoke("echo-tool", session_id, text="Async Test")
)
print(f"Request ID: {request_id}")
# Poll client.get_request_status(session_id, request_id) for the outcome.

# Stream tool execution
def stream_callback(chunk, is_final):
    if is_final:
        print("Stream completed")
    else:
        print(f"Received chunk: {chunk}")

client.stream("echo-tool", stream_callback, session_id, text="Streaming Test")

Explicit Provider Runtime with gRPC Client

from toolplane import Toolplane

client = Toolplane(
    server_host="localhost",
    server_port=9001,
    api_key="your-api-key-here",
    user_id="provider-user",
)

provider = client.provider_runtime()
session = provider.create_session(
    name="My Provider Session",
    description="Session backed by the explicit provider runtime",
    namespace="development",
)

@provider.tool(
    session_id=session.session_id,
    name="echo-tool",
    description="Echo back the input text",
    tags=["utility", "testing"],
)
def echo_tool(text: str) -> str:
    return f"Echo: {text}"

provider.run_forever()

Explicit Provider Runtime with HTTP Client

from toolplane import ToolplaneHTTP

# Initialize HTTP client against the gateway (default port 8080)
client = ToolplaneHTTP(
    server_host="localhost",
    server_port=8080,
    api_key="your-api-key-here",
)

# Create an explicit provider runtime
provider = client.provider_runtime()
session = provider.create_session(
    user_id="user123",
    name="My HTTP Provider Session",
)

# Register a tool
@provider.tool(
    session_id=session.session_id,
    name="http-tool",
    description="HTTP-based tool execution",
)
def http_tool(text: str) -> str:
    return f"HTTP Processed: {text}"

provider.run_forever()

Tools are session-scoped: a consumer can only invoke tools registered into the same session, so separate provider and consumer processes must share a session ID (this is what the examples arrange through TOOLPLANE_SESSION_ID).

Session Management

# Consumer-side session creation does not imply machine registration
session = client.create_session(
    session_id="my-unique-session-id",
    user_id="user123",
    name="My Application Session",
    description="Session for my application",
    namespace="production",
    register_machine=False,
)

# Provider-side session creation uses the explicit runtime and attaches a machine
provider = client.provider_runtime()
provider_session = provider.create_session(
    user_id="user123",
    name="Provider Session",
    namespace="production",
)

# Get an existing session
session_context = client.get_session("my-unique-session-id")

# List user sessions with pagination (page_token is the opaque cursor
# returned by the previous page; it is a string, not an offset)
user_sessions = client.list_user_sessions(
    user_id="user123",
    page_size=10,
    page_token="",
)

# Get session statistics
stats = client.get_session_stats(user_id="user123")
print(f"Total sessions: {stats['total_sessions']}")

# Bulk delete sessions
bulk_delete_result = client.bulk_delete_sessions(
    user_id="user123",
    session_ids=["session-1", "session-2"],
)

# Invalidate a session (revokes every live API key of the session)
success = client.invalidate_session("session-123", reason="Session expired")
print(f"Session invalidated: {success}")

Tool Management

# Registration uses the explicit provider runtime
provider = client.provider_runtime(["session-123"])
provider.attach_session("session-123", register_machine=True)

# Register a tool with automatic schema generation
@provider.tool(
    session_id="session-123",
    name="process-data",
    description="Process and transform data",
    tags=["data-processing", "utility"],
)
def process_data(input: str, format: str = "json") -> dict:
    return {"input": input, "format": format, "processed": True}

# Register a streaming tool
@provider.tool(
    session_id="session-123",
    name="stream-data",
    description="Stream data processing",
    stream=True,
)
def stream_data(input: str):
    for char in input:
        yield f"Processing: {char}"

# Discover tools in the session
available_tools = client.get_available_tools("session-123")
print(f"Available tools: {available_tools}")

Machine Management

# Consumer sessions are not machine-backed by default; attach through
# the provider runtime to register a machine
provider = client.provider_runtime(["session-123"])
session_context = provider.attach_session("session-123", register_machine=True)
machine_id = session_context.machine_id
print(f"Registered machine ID: {machine_id}")

# Public machine-management helpers stay on the client
machine_status = client.get_machine("session-123", machine_id)
print(f"Machine status: {machine_status}")

# Drain the machine gracefully before shutdown, then unregister
client.drain_machine("session-123", machine_id)
client.unregister_machine("session-123", machine_id)

Heartbeats are owned by the explicit provider runtime (heartbeat_interval on the client constructor configures the cadence); you never send them yourself.

Request Processing

# Create a request directly (the same path invoke() uses)
request_id = client.create_request(
    session_id="session-123",
    tool_name="process-data",
    input_data='{"input": "test", "format": "json"}',
)

# Poll request status: arguments are (session_id, request_id)
status = client.get_request_status("session-123", request_id)
print(f"Request status: {status['status']}")

# List requests in a session; page_token is the cursor from the previous page
requests = client.list_requests("session-123", status="done", limit=10, page_token="")

# Cancel a request
client.cancel_request("session-123", request_id)

Configuration Parameters

Toolplane (gRPC facade)

ToolplaneHTTP (HTTP gateway facade)

Configuration commonly comes from the environment; the examples read TOOLPLANE_SERVER_HOST, TOOLPLANE_SERVER_PORT, TOOLPLANE_API_KEY, TOOLPLANE_USER_ID, TOOLPLANE_USE_TLS, TOOLPLANE_TLS_CA_CERT_PATH, and TOOLPLANE_TLS_SERVER_NAME, defaulting the port to 9001 and the API key to toolplane-conformance-fixture-key.

Error Handling

The client raises a typed hierarchy rooted at ToolplaneError:

from toolplane import ToolplaneError, ToolplaneNotFoundError

try:
    client.invoke("missing-tool", session_id)
except ToolplaneNotFoundError as e:
    print(f"Tool not found: {e}")
except ToolplaneError as e:
    print(f"Toolplane error: {e}")

API Reference

Toolplane

Public methods parsed from toolplane/toolplane_client.py:

ToolplaneHTTP

Public methods parsed from toolplane/toolplane_http_client.py:

ProviderRuntime

Public methods parsed from toolplane/provider_runtime.py:

Contributing

Contributions are welcome — see the repository-level README Contributing section for the required checks. Run the Python suites locally with:

cd server
make python-unit            # unit tests
make conformance-python     # shared-fixture conformance

Support

• Advanced API reference: MANUAL.md
• Example walkthrough: README_EXAMPLES.md
• Package architecture: ARCHITECTURE.md
• Issues: open one on the GitHub repository