reflection.py•5.09 kB
"""enhanced reflection engine with allow/deny lists and safe defaults."""
from __future__ import annotations
import fnmatch
import inspect
from typing import Any, Callable, Dict, List, Optional, Set
from dataclasses import dataclass
@dataclass
class ToolMetadata:
"""metadata for a discovered sdk tool."""
fq_name: str # fully qualified name like "kubernetes.CoreV1Api.list_namespace"
callable_obj: Callable
signature: inspect.Signature
docstring: Optional[str]
is_mutating: bool # detected as destructive/write operation
is_async: bool
module: str
class_name: Optional[str]
class ReflectionEngine:
"""discovers and filters sdk tools with allow/deny lists."""
# common destructive operation patterns
MUTATING_PATTERNS = [
"*delete*", "*remove*", "*destroy*", "*drop*",
"*create*", "*update*", "*patch*", "*put*", "*post*",
"*set*", "*modify*", "*change*", "*write*", "*insert*"
]
# conservative default: only expose read-only operations
DEFAULT_DENY_PATTERNS = MUTATING_PATTERNS
def __init__(
self,
allow_patterns: Optional[List[str]] = None,
deny_patterns: Optional[List[str]] = None,
allow_mutating: bool = False
):
"""
initialize reflection engine.
args:
allow_patterns: glob patterns to include (e.g. ["kubernetes.CoreV1Api.*"])
deny_patterns: glob patterns to exclude (e.g. ["*.delete*"])
allow_mutating: if false, block all mutating operations by default
"""
self.allow_patterns = allow_patterns or ["*"]
self.deny_patterns = deny_patterns or []
if not allow_mutating:
self.deny_patterns.extend(self.DEFAULT_DENY_PATTERNS)
def discover_tools(self, module: Any, module_name: str) -> List[ToolMetadata]:
"""discover all callable tools in a module."""
tools: List[ToolMetadata] = []
# discover module-level functions
for name, obj in inspect.getmembers(module, inspect.isfunction):
if name.startswith("_"):
continue
fq_name = f"{module_name}.{name}"
if self._should_include(fq_name):
tools.append(self._create_metadata(fq_name, obj, module_name, None))
# discover classes and their methods
for class_name, cls in inspect.getmembers(module, inspect.isclass):
if class_name.startswith("_"):
continue
for method_name, method in inspect.getmembers(cls, inspect.isfunction):
if method_name.startswith("_"):
continue
fq_name = f"{module_name}.{class_name}.{method_name}"
if self._should_include(fq_name):
tools.append(self._create_metadata(fq_name, method, module_name, class_name))
return tools
def _create_metadata(
self,
fq_name: str,
callable_obj: Callable,
module: str,
class_name: Optional[str]
) -> ToolMetadata:
"""create metadata for a discovered tool."""
try:
sig = inspect.signature(callable_obj)
except (ValueError, TypeError):
# fallback for built-ins
sig = None
is_mutating = self._detect_mutating(fq_name, callable_obj)
is_async = inspect.iscoroutinefunction(callable_obj)
docstring = inspect.getdoc(callable_obj)
return ToolMetadata(
fq_name=fq_name,
callable_obj=callable_obj,
signature=sig,
docstring=docstring,
is_mutating=is_mutating,
is_async=is_async,
module=module,
class_name=class_name
)
def _should_include(self, fq_name: str) -> bool:
"""check if a tool should be included based on allow/deny patterns."""
# check deny list first
for pattern in self.deny_patterns:
if fnmatch.fnmatch(fq_name.lower(), pattern.lower()):
return False
# check allow list
for pattern in self.allow_patterns:
if fnmatch.fnmatch(fq_name.lower(), pattern.lower()):
return True
return False
def _detect_mutating(self, fq_name: str, callable_obj: Callable) -> bool:
"""detect if a method is mutating/destructive."""
name_lower = fq_name.lower()
# check against mutating patterns
for pattern in self.MUTATING_PATTERNS:
if fnmatch.fnmatch(name_lower, pattern):
return True
# check docstring for hints
doc = inspect.getdoc(callable_obj)
if doc:
doc_lower = doc.lower()
if any(word in doc_lower for word in [
"delete", "remove", "destroy", "create", "update",
"modify", "write", "insert", "patch"
]):
return True
return False