Tool Schema Design

tools = [
    {
        "type": "function",
        "function": {
            "name": "kubectl_get",
            "description": "Get Kubernetes resources. Returns current state of pods, services, deployments, or other resources.",
            "parameters": {
                "type": "object",
                "properties": {
                    "resource": {
                        "type": "string",
                        "description": "Kubernetes resource type (pods, services, deployments, nodes, etc.)",
                        "enum": ["pods", "services", "deployments", "nodes", "configmaps", "secrets", "ingress"]
                    },
                    "namespace": {
                        "type": "string",
                        "description": "Kubernetes namespace. Defaults to 'default' if not specified."
                    },
                    "label_selector": {
                        "type": "string",
                        "description": "Filter by labels, e.g., 'app=checkout' or 'env=production'"
                    },
                    "field_selector": {
                        "type": "string",
                        "description": "Filter by fields, e.g., 'status.phase=Running'"
                    }
                },
                "required": ["resource"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_prometheus",
            "description": "Execute a PromQL query against Prometheus and return time-series results.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Valid PromQL query, e.g., 'rate(http_requests_total[5m])'"
                    },
                    "start": {
                        "type": "string",
                        "description": "Start time in ISO format or relative like '1h ago'"
                    },
                    "end": {
                        "type": "string",
                        "description": "End time in ISO format or 'now'"
                    },
                    "step": {
                        "type": "string",
                        "description": "Query resolution step, e.g., '15s', '1m'",
                        "default": "15s"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

import jsonschema

class ToolValidator:
    def __init__(self, tools):
        self.schemas = {
            t["function"]["name"]: t["function"]["parameters"]
            for t in tools
        }

    def validate(self, tool_name, arguments):
        """Validate tool arguments against the schema."""
        schema = self.schemas.get(tool_name)
        if not schema:
            return {"valid": False, "error": f"Unknown tool: {tool_name}"}

        try:
            jsonschema.validate(arguments, schema)
            return {"valid": True}
        except jsonschema.ValidationError as e:
            return {
                "valid": False,
                "error": f"Invalid argument '{e.path[-1]}': {e.message}",
                "expected_type": schema["properties"].get(str(e.path[-1]), {}).get("type"),
                "allowed_values": schema["properties"].get(str(e.path[-1]), {}).get("enum"),
            }

# Usage
validator = ToolValidator(tools)
result = validator.validate("kubectl_get", {"resource": "pods", "namespace": "production"})
if not result["valid"]:
    print(f"Validation failed: {result['error']}")

def execute_tool_with_feedback(tool_name, arguments):
    """Execute a tool and return helpful errors."""
    # Validate
    validation = validator.validate(tool_name, arguments)
    if not validation["valid"]:
        # Return the error to the model so it can retry
        return {
            "error": True,
            "message": f"Tool call failed validation: {validation['error']}",
            "hint": f"Expected type: {validation.get('expected_type')}. Allowed: {validation.get('allowed_values')}"
        }

    # Execute
    try:
        result = tools[tool_name].execute(**arguments)
        return {"error": False, "result": result}
    except ConnectionError as e:
        return {
            "error": True,
            "message": f"Connection failed: {e}",
            "hint": "Check that the service is running and accessible"
        }
    except PermissionError as e:
        return {
            "error": True,
            "message": f"Permission denied: {e}",
            "hint": "The service account may lack required RBAC permissions"
        }

class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def register(self, name, fn, description, parameters):
        """Register a new tool."""
        self.tools[name] = {
            "function": fn,
            "schema": {
                "type": "function",
                "function": {
                    "name": name,
                    "description": description,
                    "parameters": parameters
                }
            }
        }

    def get_schemas(self):
        """Get all tool schemas for the LLM."""
        return [t["schema"] for t in self.tools.values()]

    def execute(self, name, arguments):
        """Execute a tool by name."""
        tool = self.tools.get(name)
        if not tool:
            raise ValueError(f"Unknown tool: {name}")
        return tool["function"](**arguments)

# Register tools
registry = ToolRegistry()
registry.register("kubectl_get", kubectl_get_fn, "Get K8s resources", kubectl_get_schema)
registry.register("query_prometheus", prom_query_fn, "Query Prometheus", prom_schema)