"""
Auto-discovery system for MCP tools.
This module automatically discovers and registers all tool modules
in this directory and subdirectories. Any module with a `register(mcp)`
function will be automatically loaded.
Benefits:
- No manual registration needed
- Easy to add new tools (just create a file with register())
- Supports nested packages
- Clean separation of concerns
How to add a new tool:
1. Create a new file or package under tools/
2. Define a `register(mcp)` function
3. That's it! Auto-discovery handles the rest.
Example:
```python
# tools/my_tool.py
from mcp.server.fastmcp import FastMCP
def register_all(registry: ContextAwareToolRegistry) -> None:
@registry.tool()
async def my_tool(param: str):
'''Tool description.'''
return {"result": param}
```
"""
from __future__ import annotations
import importlib
import pkgutil
from types import ModuleType
from typing import Callable
from mcp.server.fastmcp import FastMCP
from src.observability import get_logger
logger = get_logger(__name__)
def register_all(registry: ContextAwareToolRegistry) -> None:
"""
Discover and register all tools in this package.
This walks through all modules in the tools/ directory and its
subdirectories, looking for modules that expose a `register_all(registry)` function.
Args:
registry: Context-aware tool registry to register tools with
"""
pkg = __name__
registered_count = 0
failed_count = 0
logger.info("tool_discovery_started", package=pkg)
# Walk through all modules in this package
for _, modname, _ in pkgutil.walk_packages(__path__, prefix=pkg + "."):
try:
# Import the module
mod: ModuleType = importlib.import_module(modname)
# Check if it has a register function
register_func: Callable | None = getattr(mod, "register", None)
logger.info(
"tool_module_discovered",
module=modname,
has_register=bool(register_func),
)
if callable(register_func):
# Call the register function
register_func(registry)
registered_count += 1
logger.info(
"tool_registered",
module=modname,
success=True,
)
else:
# Module doesn't expose a register() function; skip it
logger.info(
"tool_module_skipped_no_register",
module=modname,
)
except Exception as e:
failed_count += 1
logger.error(
"tool_registration_failed",
module=modname,
error=str(e),
error_type=type(e).__name__
)
# Continue with other modules even if one fails
continue
logger.info(
"tool_discovery_completed",
registered=registered_count,
failed=failed_count,
total=registered_count + failed_count
)
