fix(hanzo-mcp): add missing mypy type annotations to pass CI

This commit is contained in:
2026-03-01 16:08:00 -08:00
parent 3e3e1ac861
commit 9f0d4e7a97
3 changed files with 24 additions and 13 deletions
+5 -5
View File
@@ -79,12 +79,12 @@ class MemoryBackendPlugin(Protocol):
class PluginRegistry:
"""Registry for memory backend plugins."""
def __init__(self):
def __init__(self) -> None:
self._plugins: Dict[str, MemoryBackendPlugin] = {}
self._active_plugins: List[str] = []
self._initialized = False
self._initialized: bool = False
def register_plugin(self, plugin: MemoryBackendPlugin):
def register_plugin(self, plugin: MemoryBackendPlugin) -> None:
"""Register a new plugin."""
self._plugins[plugin.name] = plugin
@@ -136,14 +136,14 @@ class PluginRegistry:
result.append(name)
return result
async def initialize_all_active(self):
async def initialize_all_active(self) -> None:
"""Initialize all active plugins."""
if not self._initialized:
for plugin in self.get_active_plugins():
await plugin.initialize()
self._initialized = True
async def shutdown_all_active(self):
async def shutdown_all_active(self) -> None:
"""Shutdown all active plugins."""
for plugin in self.get_active_plugins():
await plugin.shutdown()
@@ -1,17 +1,23 @@
"""Test helper classes for MCP tools testing."""
from typing import Any, Dict
from typing import Any, Dict, List, Optional
class PaginatedResponseWrapper:
"""Wrapper class for paginated responses to support tests."""
def __init__(self, items=None, next_cursor=None, has_more=False, total_items=None):
def __init__(
self,
items: Optional[List[Any]] = None,
next_cursor: Optional[str] = None,
has_more: bool = False,
total_items: Optional[int] = None,
) -> None:
"""Initialize paginated response."""
self.items = items or []
self.next_cursor = next_cursor
self.has_more = has_more
self.total_items = total_items or len(self.items)
self.items: List[Any] = items or []
self.next_cursor: Optional[str] = next_cursor
self.has_more: bool = has_more
self.total_items: int = total_items or len(self.items)
def to_json(self) -> Dict[str, Any]:
"""Convert to JSON-serializable dict."""
@@ -26,6 +32,11 @@ class PaginatedResponseWrapper:
# Export a convenience constructor
def PaginatedResponse(items=None, next_cursor=None, has_more=False, total_items=None):
def PaginatedResponse(
items: Optional[List[Any]] = None,
next_cursor: Optional[str] = None,
has_more: bool = False,
total_items: Optional[int] = None,
) -> PaginatedResponseWrapper:
"""Create a paginated response for testing."""
return PaginatedResponseWrapper(items, next_cursor, has_more, total_items)
@@ -89,7 +89,7 @@ def format_timeout(seconds: float) -> str:
# Test the parser
if __name__ == "__main__":
test_cases = [
test_cases: list[Union[str, int, float]] = [
"2min",
"5m",
"120s",