mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
fix(hanzo-mcp): add missing mypy type annotations to pass CI
This commit is contained in:
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user