Ryan Malloy
31948d6ffc
Some checks are pending
Test Dashboard / test-and-dashboard (push) Waiting to run
Named for Milton Waddams, who was relocated to the basement with boxes of legacy documents. He handles the .doc and .xls files from 1997 that nobody else wants to touch. - Rename package from mcp-office-tools to mcwaddams - Update author to Ryan Malloy - Update all imports and references - Add Office Space themed README narrative - All 53 tests passing
103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
"""
|
|
Decorators for MCP Office Tools.
|
|
|
|
Provides common patterns for error handling and Pydantic field resolution.
|
|
"""
|
|
|
|
from functools import wraps
|
|
from typing import Any, Callable, TypeVar
|
|
|
|
from pydantic.fields import FieldInfo
|
|
|
|
from .validation import OfficeFileError
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
def resolve_field_defaults(**defaults: Any) -> Callable:
|
|
"""
|
|
Decorator to resolve Pydantic Field defaults for direct function calls.
|
|
|
|
When MCP tool methods are called directly (outside the MCP framework),
|
|
Pydantic Field() defaults aren't automatically applied - parameters
|
|
remain as FieldInfo objects. This decorator converts them to actual values.
|
|
|
|
Usage:
|
|
@mcp_tool(...)
|
|
@resolve_field_defaults(sheet_names=[], include_statistics=True)
|
|
async def analyze_excel_data(self, file_path: str, sheet_names: list = Field(...)):
|
|
# sheet_names will be [] if called directly without argument
|
|
...
|
|
|
|
Args:
|
|
**defaults: Mapping of parameter names to their default values
|
|
|
|
Returns:
|
|
Decorated async function with resolved defaults
|
|
"""
|
|
import inspect
|
|
|
|
def decorator(func: Callable[..., T]) -> Callable[..., T]:
|
|
sig = inspect.signature(func)
|
|
param_names = list(sig.parameters.keys())
|
|
|
|
@wraps(func)
|
|
async def wrapper(self, *args, **kwargs):
|
|
# Build a dict of all parameter values (combining args and kwargs)
|
|
# Skip 'self' which is the first parameter
|
|
bound_args = {}
|
|
for i, arg in enumerate(args):
|
|
if i + 1 < len(param_names): # +1 to skip 'self'
|
|
bound_args[param_names[i + 1]] = arg
|
|
|
|
# Merge with kwargs
|
|
bound_args.update(kwargs)
|
|
|
|
# For parameters not provided, check if default is FieldInfo
|
|
for param_name, default_value in defaults.items():
|
|
if param_name not in bound_args:
|
|
# Parameter using its default value - set to our resolved default
|
|
kwargs[param_name] = default_value
|
|
elif isinstance(bound_args[param_name], FieldInfo):
|
|
# Explicitly passed FieldInfo - resolve it
|
|
kwargs[param_name] = default_value
|
|
|
|
return await func(self, *args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def handle_office_errors(operation_name: str) -> Callable:
|
|
"""
|
|
Decorator for consistent error handling in Office document operations.
|
|
|
|
Wraps async functions to catch exceptions and re-raise them as
|
|
OfficeFileError with a descriptive message. Already-raised
|
|
OfficeFileError exceptions are passed through unchanged.
|
|
|
|
Usage:
|
|
@mcp_tool(...)
|
|
@handle_office_errors("Excel analysis")
|
|
async def analyze_excel_data(self, file_path: str):
|
|
# Any exception becomes: OfficeFileError("Excel analysis failed: ...")
|
|
...
|
|
|
|
Args:
|
|
operation_name: Human-readable name for the operation (used in error messages)
|
|
|
|
Returns:
|
|
Decorated async function with error handling
|
|
"""
|
|
def decorator(func: Callable[..., T]) -> Callable[..., T]:
|
|
@wraps(func)
|
|
async def wrapper(*args, **kwargs):
|
|
try:
|
|
return await func(*args, **kwargs)
|
|
except OfficeFileError:
|
|
# Re-raise our custom errors unchanged
|
|
raise
|
|
except Exception as e:
|
|
raise OfficeFileError(f"{operation_name} failed: {str(e)}")
|
|
return wrapper
|
|
return decorator
|