Spaces:
Sleeping
Sleeping
File size: 3,081 Bytes
f2b5c2a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """
Base class for MCP tools with validation.
All MCP tools should inherit from this base class to ensure
consistent validation and error handling.
"""
import logging
from typing import Dict, Any
from pydantic import BaseModel, Field
from abc import ABC, abstractmethod
logger = logging.getLogger(__name__)
class ToolResult(BaseModel):
"""Standard result format for all MCP tools."""
success: bool
message: str
data: Dict[str, Any] = Field(default_factory=dict)
class MCPToolBase(ABC):
"""
Abstract base class for MCP tools.
All tools must implement the execute method and define their schema.
"""
@property
@abstractmethod
def name(self) -> str:
"""Tool name (must be unique)."""
pass
@property
@abstractmethod
def description(self) -> str:
"""Human-readable description of what the tool does."""
pass
@property
@abstractmethod
def parameters(self) -> Dict[str, Any]:
"""
JSON schema describing the tool's parameters.
Format for Cohere API:
{
"param_name": {
"description": "Parameter description",
"type": "string|integer|boolean",
"required": True|False
}
}
"""
pass
@abstractmethod
async def execute(self, **kwargs) -> Dict[str, Any]:
"""
Execute the tool with given parameters.
Args:
**kwargs: Tool parameters
Returns:
Dictionary with tool execution result
Raises:
ValueError: If parameters are invalid
Exception: If execution fails
"""
pass
def validate_user_id(self, user_id: int) -> None:
"""
Validate that user_id is provided and valid.
Args:
user_id: User ID to validate
Raises:
ValueError: If user_id is invalid
"""
if not user_id or user_id <= 0:
raise ValueError("Invalid user_id")
def create_success_result(
self,
message: str,
data: Dict[str, Any] = None
) -> Dict[str, Any]:
"""
Create a standardized success result.
Args:
message: Success message
data: Optional data to include
Returns:
Standardized success result dictionary
"""
result = ToolResult(
success=True,
message=message,
data=data or {}
)
return result.model_dump()
def create_error_result(
self,
message: str,
error_code: str = "TOOL_ERROR"
) -> Dict[str, Any]:
"""
Create a standardized error result.
Args:
message: Error message
error_code: Error code for categorization
Returns:
Standardized error result dictionary
"""
return {
"success": False,
"message": message,
"error_code": error_code,
"data": {}
}
|