RPC System (System Tools)¶

System tools are Python functions that run on a specific cluster node, usually because they need host-level access such as hardware, the OS, or local files. They are exposed over gRPC and called through a central gateway.

The system has three parts:

Part	Package / service	Role
Tools RPC SDK	`tools_rpc_sdk` (`src/tools/rpc_sdk`)	Define tools with `@register_tool` and serve them over gRPC on each node (port `50051`)
Org RPC Executor	`src/tools/org_rpc_executor`	Stores which tool lives on which node (MongoDB), resolves the node IP via Kubernetes, and forwards calls
Kubernetes node labels	`nodeID=<node_id>`	Map a tool's `node_id` to a real node

org-tools-proxy

client ──POST /cluster-tools/execute──> Org RPC Executor
                                          │ 1. find (tool_name, node_id) in MongoDB
                                          │ 2. find k8s node with label nodeID=<node_id> → InternalIP
                                          ▼
                                 <InternalIP>:50051  (gRPC ToolExecutor.CallFunction)
                                          │
                                   your @register_tool function

1. Write and serve system tools¶

Install the SDK¶

cd functions-tools/src/tools/rpc_sdk
touch README.md          # setup.py reads README.md; the file is not in the repo yet
pip install .

Define tools¶

A system tool takes bytes and returns bytes. JSON is the usual encoding.

# my_tools.py
import json
import platform
from tools_rpc_sdk.decorator import register_tool

NODE_ID = "node-abc123"   # must match the nodeID label on this node

@register_tool(
    name="reverse_text",
    input_spec={"text": "string"},
    description="Reverses the 'text' field of a JSON payload",
    metadata={"node_id": NODE_ID, "version": "1.0.0"},
)
def reverse_text(input_bytes: bytes) -> bytes:
    data = json.loads(input_bytes)
    return json.dumps({"result": data["text"][::-1]}).encode()

@register_tool(
    name="host_info",
    description="Returns basic host information",
    metadata={"node_id": NODE_ID, "version": "1.0.0"},
)
def host_info(input_bytes: bytes) -> bytes:
    return json.dumps({"hostname": platform.node(), "os": platform.system()}).encode()

register_tool(name, input_spec=None, metadata=None, description="")

Argument	Notes
`name`	Unique per process. Registering the same name twice raises `ValueError`.
`input_spec`	Free-form description of the expected input (documentation only).
`metadata`	Put `node_id` here; it is used when syncing to the Org RPC Executor. All values must be strings, because they are returned in the gRPC `map<string,string>` metadata field.
`description`	Human-readable summary.

If the function raises, or returns something other than bytes, the call returns success=false and the error plus traceback in error.

Start the gRPC server¶

# run_server.py
import my_tools  # importing the module registers the tools
from tools_rpc_sdk.server import serve

if __name__ == "__main__":
    serve(port=50051)

export ORG_SYSTEM_TOOLS_REGISTRY_URL=http://org-rpc-executor:8080   # optional, see below
python run_server.py
# INFO:ToolGRPCServer:Starting ToolExecutor gRPC server on port 50051

On startup, if ORG_SYSTEM_TOOLS_REGISTRY_URL is set, every tool is registered with the Org RPC Executor via POST {url}/tools with this body:

{
  "name": "reverse_text",
  "node_id": "node-abc123",
  "input_spec": {"text": "string"},
  "metadata": {"node_id": "node-abc123", "version": "1.0.0"},
  "description": "Reverses the 'text' field of a JSON payload"
}

If the variable is not set, syncing is skipped and the server still runs.

Duplicate entries on restart

Before re-registering, the syncer tries to delete the old entry at /tools/<name>/<node_id>, but the executor's route is /tools/<node_id>/<name>. The delete fails silently, so each restart adds another copy of the entry. Lookups still work because the first match is used. Clean up with DELETE /tools/<node_id>/<name> if needed.

Run in Docker¶

The server needs host access, so run it with host networking, one container per node.

FROM python:3.10-slim
WORKDIR /app
COPY rpc_sdk/ /sdk/
RUN touch /sdk/README.md && pip install /sdk
COPY my_tools.py run_server.py ./
CMD ["python", "run_server.py"]

docker build -t node-tools .
docker run --net=host \
  -e ORG_SYSTEM_TOOLS_REGISTRY_URL=http://org-rpc-executor:8080 \
  node-tools

Label the Kubernetes node so the executor can find it:

kubectl label node <k8s-node-name> nodeID=node-abc123

2. Call a tool directly over gRPC¶

Protocol¶

syntax = "proto3";
package toolrpc;

service ToolExecutor {
  rpc CallFunction (CallFunctionRequest) returns (CallFunctionResponse);
}

message CallFunctionRequest {
  string tool_name = 1;
  bytes input = 2;                    // raw input bytes
}

message CallFunctionResponse {
  bool success = 1;
  bytes output = 2;                   // raw output bytes
  string error = 3;                   // error message + traceback on failure
  map<string, string> metadata = 4;   // the tool's metadata
}

The generated Python stubs ship with the SDK as tools_rpc_sdk.rpc_service_pb2 and tools_rpc_sdk.rpc_service_pb2_grpc.

Client¶

# client.py
import json
import grpc
from tools_rpc_sdk import rpc_service_pb2, rpc_service_pb2_grpc

def call_tool(tool_name, payload, target="localhost:50051"):
    with grpc.insecure_channel(target) as channel:
        stub = rpc_service_pb2_grpc.ToolExecutorStub(channel)
        resp = stub.CallFunction(rpc_service_pb2.CallFunctionRequest(
            tool_name=tool_name,
            input=json.dumps(payload).encode(),
        ))
    if not resp.success:
        raise RuntimeError(resp.error)
    return json.loads(resp.output), dict(resp.metadata)

if __name__ == "__main__":
    print(call_tool("reverse_text", {"text": "hello world"}))
    print(call_tool("host_info", {}))

({'result': 'dlrow olleh'}, {'version': '1.0.0', 'node_id': 'node-abc123'})
({'hostname': 'worker-1', 'os': 'Linux'}, {'version': '1.0.0', 'node_id': 'node-abc123'})

3. Org RPC Executor¶

The Org RPC Executor is the cluster gateway. Clients call it by tool name and node, and it locates and calls the right node.

Deployment requirements¶

Runs inside Kubernetes: it loads the in-cluster config at startup and needs RBAC permission to list nodes.
MongoDB at mongodb://localhost:27017 (currently hard-coded), database tools_db, collection cluster_local_tools. Run MongoDB as a sidecar or change ClusterLocalToolsDB defaults.
Node gRPC servers listen on port 50051.

The package provides two Flask blueprints but no entry point. Mount them in a small app:

# server.py (place next to the core/ package)
from flask import Flask
from core.apis import cluster_tools_bp, tools_executor_bp

app = Flask(__name__)
app.register_blueprint(cluster_tools_bp)
app.register_blueprint(tools_executor_bp)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

The examples below use port 8080; use whatever port you choose.

Data model¶

@dataclass
class ClusterLocalTools:
    name: str
    func: Callable[[bytes], bytes]   # runtime only; never stored
    node_id: str
    input_spec: Dict[str, Any] = field(default_factory=dict)
    metadata: Dict[str, Any] = field(default_factory=dict)
    description: str = ""

A tool is identified by the pair (name, node_id), so the same tool name can exist on several nodes.

API reference¶

Set GW=http://org-rpc-executor:8080 in the examples below.

Method	Path	Purpose
POST	`/cluster-tools/execute`	Execute a tool on a node
POST	`/cluster-tools/resolve`	Resolve a tool's gRPC endpoint (see note)
POST	`/tools`	Register a tool (used by the SDK sync)
GET	`/tools/<node_id>`	List tools on a node
GET	`/tools/<node_id>/<name>`	Get one tool
PUT	`/tools/<node_id>/<name>`	Update fields
DELETE	`/tools/<node_id>/<name>`	Delete

POST `/cluster-tools/execute`¶

All three fields are required.

curl -X POST $GW/cluster-tools/execute \
  -H "Content-Type: application/json" \
  -d '{"tool_name": "reverse_text", "node_id": "node-abc123", "input": {"text": "hello world"}}'

input is JSON-encoded to bytes before being sent to the tool. If the tool's output is valid JSON it is decoded, otherwise it is returned as a string.

{
  "success": true,
  "output": {"result": "dlrow olleh"},
  "error": "",
  "metadata": {"node_id": "node-abc123", "version": "1.0.0"}
}

Status	When
200	Tool ran successfully
400	Missing body, or missing `tool_name` / `node_id` / `input`
500	Tool not registered for that node, node not found in Kubernetes, gRPC failure, or the tool raised. `success` is `false` and `error` explains why.

POST `/cluster-tools/resolve`¶

curl -X POST $GW/cluster-tools/resolve \
  -H "Content-Type: application/json" -d '{"tool_name": "reverse_text"}'

Intended response: {"success": true, "endpoint": "10.42.0.12:50051"}.

Known issue

In the current build this endpoint always returns 500, because it calls the DB lookup with a wrong keyword argument (tool_name= instead of name=). Use /cluster-tools/execute with an explicit node_id, or list a node's tools with GET /tools/<node_id>.

Registry routes¶

# register manually (same body the SDK sync sends)
curl -X POST $GW/tools -H "Content-Type: application/json" -d '{
  "name": "reverse_text", "node_id": "node-abc123",
  "description": "Reverses text", "input_spec": {"text": "string"},
  "metadata": {"version": "1.0.0"}
}'
# -> 201 {"message": "Tool created", "id": "<mongo id>"}

curl $GW/tools/node-abc123                    # list tools on node
curl $GW/tools/node-abc123/reverse_text       # one tool
curl -X PUT $GW/tools/node-abc123/reverse_text \
  -H "Content-Type: application/json" -d '{"description": "Reverses a string"}'
curl -X DELETE $GW/tools/node-abc123/reverse_text

POST /tools requires name, node_id, and description.

How resolution works¶

If node_id is already cached, use the cached IP and skip the remaining steps.
Otherwise, look up (tool_name, node_id) in cluster_local_tools. If it isn't there, the call fails.
List Kubernetes nodes and pick the one labelled nodeID=<node_id>.
Take that node's InternalIP, cache node_id → ip, and call <ip>:50051.

The cache is never invalidated, so restart the executor if a node's IP changes. Because cached nodes skip step 2, a call for an unregistered tool on a cached node reaches the node, which then returns success=false with Tool '<name>' not found.