Tool Plugin — Python

A complete example of a remote tool plugin in Python that provides a get_weather tool.

Project Setup

mkdir weather-plugin && cd weather-plugin
python -m venv venv && source venv/bin/activate
pip install grpcio grpcio-tools

Generate Python code from the proto files:

python -m grpc_tools.protoc \
  --proto_path=. \
  --python_out=. \
  --grpc_python_out=. \
  proto/plugin.proto proto/tool.proto

This generates plugin_pb2.py, plugin_pb2_grpc.py, tool_pb2.py, and tool_pb2_grpc.py.

Full Source — `main.py`

import json
import os
from concurrent import futures

import grpc

import plugin_pb2
import plugin_pb2_grpc
import tool_pb2
import tool_pb2_grpc


MANIFEST_HCL = """plugin "weather" {
  version     = "1.0.0"
  description = "Provides current weather data for any city."
  author      = "yourorg"
  icon        = "cloud"
  type        = "remote"

  config_schema {
    field "api_key_env" {
      type        = "string"
      required    = true
      placeholder = "e.g. WEATHER_API_KEY"
      description = "Environment variable containing your weather API key"
    }

    field "unit" {
      type    = "string"
      options = ["fahrenheit", "celsius"]
      default = "fahrenheit"
      description = "Temperature unit"
    }

    field "max_retries" {
      type    = "number"
      default = "3"
      description = "Maximum retry attempts for API calls"
    }

    field "debug" {
      type    = "bool"
      default = "false"
      description = "Enable debug logging"
    }
  }
}
"""


class WeatherPlugin(
    plugin_pb2_grpc.PluginServicer,
    tool_pb2_grpc.ToolProviderServicer,
):
    """Implements both Plugin and ToolProvider services."""

    # --- Plugin service (required) ---

    def Register(self, request, context):
        return plugin_pb2.RegisterResponse(
            name="weather",
            capabilities=[plugin_pb2.TOOL_PROVIDER],
            metadata={"version": "1.0.0"},
        )

    def Health(self, request, context):
        return plugin_pb2.HealthResponse(status=plugin_pb2.SERVING)

    def GetManifest(self, request, context):
        return plugin_pb2.GetManifestResponse(manifest_hcl=MANIFEST_HCL)

    # --- ToolProvider service ---

    def ListTools(self, request, context):
        return tool_pb2.ListToolsResponse(
            tools=[
                tool_pb2.ToolDefinition(
                    name="get_weather",
                    description="Get the current weather for a city.",
                    parameters_json_schema=json.dumps({
                        "type": "object",
                        "properties": {
                            "city": {
                                "type": "string",
                                "description": "City name (e.g. 'San Francisco')",
                            }
                        },
                        "required": ["city"],
                    }),
                )
            ]
        )

    def ExecuteTool(self, request, context):
        if request.name != "get_weather":
            return tool_pb2.ExecuteToolResponse(
                result_json=json.dumps({"error": f"unknown tool: {request.name}"}),
                is_error=True,
            )

        args = json.loads(request.arguments_json)
        city = args.get("city", "Unknown")

        # In a real plugin you'd call a weather API here.
        result = {
            "city": city,
            "temperature": 72,
            "unit": "F",
            "condition": "Sunny",
        }

        return tool_pb2.ExecuteToolResponse(
            result_json=json.dumps(result),
            is_error=False,
        )


def serve():
    port = os.environ.get("PORT", "9010")
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    servicer = WeatherPlugin()
    plugin_pb2_grpc.add_PluginServicer_to_server(servicer, server)
    tool_pb2_grpc.add_ToolProviderServicer_to_server(servicer, server)
    server.add_insecure_port(f"[::]:{port}")
    server.start()
    print(f"weather-plugin listening on :{port}")
    server.wait_for_termination()


if __name__ == "__main__":
    serve()

Run It

python main.py
# weather-plugin listening on :9010

Register in Workspace

Add to your workspace HCL config:

workspace "my-workspace" {
  plugin "weather" {
    source  = "remote://localhost:9010"
    version = "1.0.0"
  }
}

Start the workspace. The agent now has access to the get_weather tool.

How It Works