refactor(tools): Refactor tool provider system for MCP support

- Introduce Tool model combining schema, metadata, and invoke callable
- Move ToolCapability, ToolMetadata, and ToolDefinition to tools/models.py
- Add automatic wiring in ToolProvider.tools() that binds tool names to methods
- Update all providers to use create_tool_schemas() + method pattern
- Standardize provider_name as @Property across all providers
- Rename redis_cli provider to redis_command
- Update documentation and tests to match new patterns
- Fix pre-existing linting issues (N806, N814, E402)

Author: abrookins

docs/how-to/tool-providers.md

@@ -14,8 +14,8 @@ This is an early release with an initial set of built-in providers. More provide
   - Config: `TOOLS_PROMETHEUS_URL`, `TOOLS_PROMETHEUS_DISABLE_SSL`
 - **Loki logs**: `redis_sre_agent.tools.logs.loki.provider.LokiToolProvider`
   - Config: `TOOLS_LOKI_URL`, `TOOLS_LOKI_TENANT_ID`, `TOOLS_LOKI_TIMEOUT`
-- **Redis CLI diagnostics**: `redis_sre_agent.tools.diagnostics.redis_cli.provider.RedisCliToolProvider`
-  - Runs Redis CLI commands against target instances
+- **Redis command diagnostics**: `redis_sre_agent.tools.diagnostics.redis_command.provider.RedisCommandToolProvider`
+  - Runs Redis commands against target instances
 - **Host telemetry**: `redis_sre_agent.tools.host_telemetry.provider.HostTelemetryToolProvider`
   - System-level metrics and diagnostics
 
@@ -41,18 +41,22 @@ Implement a ToolProvider subclass that defines tool schemas and resolves calls.
 ### Minimal skeleton
 
 ```python
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List
 from redis_sre_agent.tools.protocols import ToolProvider
-from redis_sre_agent.tools.tool_definition import ToolDefinition
+from redis_sre_agent.tools.models import ToolDefinition, ToolCapability
+
 
 class MyMetricsProvider(ToolProvider):
-    provider_name = "my_metrics"
+    @property
+    def provider_name(self) -> str:
+        return "my_metrics"
 
     def create_tool_schemas(self) -> List[ToolDefinition]:
         return [
             ToolDefinition(
                 name=self._make_tool_name("query"),
                 description="Query my metrics backend using a query string.",
+                capability=ToolCapability.METRICS,
                 parameters={
                     "type": "object",
                     "properties": {"query": {"type": "string"}},
@@ -61,17 +65,13 @@ class MyMetricsProvider(ToolProvider):
             )
         ]
 
-    async def resolve_tool_call(self, tool_name: str, args: Dict[str, Any]):
-        op = self.resolve_operation(tool_name)
-        if op == "query":
-            return await self.query(**args)
-        raise ValueError(f"Unknown operation: {op}")
-
     async def query(self, query: str) -> Dict[str, Any]:
         # Implement your backend call
         return {"status": "success", "query": query, "data": []}
 ```
 
+The base class `tools()` method automatically wires tool names to provider methods. When an LLM invokes `my_metrics_{hash}_query`, the framework calls `self.query(**args)` directly. No manual `resolve_tool_call()` implementation is required.
+
 ### Register your provider
 
 - Install your package into the same environment as the agent (e.g., `pip install -e /path/to/pkg`)

pyproject.toml

@@ -59,6 +59,26 @@ dependencies = [
     "opentelemetry-instrumentation-openai>=0.47.5",
 ]
 
+[dependency-groups]
+dev = [
+    "pytest>=7.0.0",
+    "pytest-asyncio>=0.21.0",
+    "pytest-cov>=4.1.0",
+    "ruff>=0.3.0",
+    "black>=23.0.0",
+    "mypy>=1.8.0",
+    "testcontainers>=3.7.0",
+    "pre-commit>=3.6.0",
+    "safety>=3.0.0",
+    "bandit>=1.7.0",
+    # OpenAPI client generator
+    "openapi-python-client>=0.21.0",
+    "mkdocs>=1.6.1",
+    "mkdocs-material>=9.6.22",
+    "mfcqi>=0.0.4",
+]
+
+
 [project.scripts]
 redis-sre-agent = "redis_sre_agent.cli:main"
 
@@ -77,22 +97,6 @@ include = [
 
 [tool.uv]
 default-groups = []
-dev-dependencies = [
-    "pytest>=7.0.0",
-    "pytest-asyncio>=0.21.0",
-    "pytest-cov>=4.1.0",
-    "ruff>=0.3.0",
-    "black>=23.0.0",
-    "mypy>=1.8.0",
-    "testcontainers>=3.7.0",
-    "pre-commit>=3.6.0",
-    "safety>=3.0.0",
-    "bandit>=1.7.0",
-    # OpenAPI client generator
-    "openapi-python-client>=0.21.0",
-    "mkdocs>=1.6.1",
-    "mkdocs-material>=9.6.22",
-]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

redis_sre_agent/agent/helpers.py

@@ -85,23 +85,23 @@ def sanitize_messages_for_llm(msgs: List[Any]) -> List[Any]:
     for m in msgs:
         if isinstance(m, _AI):
             try:
-                for tc in getattr(m, "tool_calls", []) or []:
+                for tc in m.tool_calls or []:
                     if isinstance(tc, dict):
                         tid = tc.get("id") or tc.get("tool_call_id")
                         if tid:
                             seen_tool_ids.add(tid)
             except Exception:
                 pass
             clean.append(m)
-        elif isinstance(m, _TM) or getattr(m, "type", "") == "tool":
-            tid = getattr(m, "tool_call_id", None)
+        elif isinstance(m, _TM) or m.type == "tool":
+            tid = m.tool_call_id
             if tid and tid in seen_tool_ids:
                 clean.append(m)
             else:
                 continue
         else:
             clean.append(m)
-    while clean and (isinstance(clean[0], _TM) or getattr(clean[0], "type", "") == "tool"):
+    while clean and (isinstance(clean[0], _TM) or clean[0].type == "tool"):
         clean = clean[1:]
     return clean
 
@@ -122,26 +122,26 @@ def _compact_messages_tail(msgs: List[Any], limit: int = 6) -> List[Dict[str, An
     tail = msgs[-limit:] if msgs else []
     compact: List[Dict[str, Any]] = []
     for m in tail:
-        role = getattr(m, "type", m.__class__.__name__.lower())
+        role = m.type if m.type else m.__class__.__name__.lower()
         row: Dict[str, Any] = {"role": role}
         try:
-            is_ai = (_AI is not None and isinstance(m, _AI)) or getattr(m, "type", "") in (
+            is_ai = (_AI is not None and isinstance(m, _AI)) or m.type in (
                 "ai",
                 "assistant",
             )
             if is_ai:
                 ids: List[str] = []
-                for tc in getattr(m, "tool_calls", []) or []:
+                for tc in m.tool_calls or []:
                     if isinstance(tc, dict):
                         tid = tc.get("id") or tc.get("tool_call_id")
                         if tid:
                             ids.append(tid)
                 if ids:
                     row["tool_calls"] = ids
-            is_tool = (_TM is not None and isinstance(m, _TM)) or getattr(m, "type", "") == "tool"
+            is_tool = (_TM is not None and isinstance(m, _TM)) or m.type == "tool"
             if is_tool:
-                row["tool_call_id"] = getattr(m, "tool_call_id", None)
-                name = getattr(m, "name", None)
+                row["tool_call_id"] = m.tool_call_id
+                name = m.name
                 if name:
                     row["name"] = name
         except Exception:
@@ -193,7 +193,7 @@ def build_result_envelope(
 
     from .models import ResultEnvelope
 
-    content = getattr(tool_message, "content", None)
+    content = tool_message.content
     data_obj = None
     if isinstance(content, str) and content:
         try:
@@ -210,9 +210,8 @@ def _extract_operation_from_tool_name(full: str) -> str:
         parts = full.split(".")
         return parts[-1] if parts else full
 
-    description = (
-        getattr(tooldefs_by_name.get(tool_name), "description", None) if tool_name else None
-    )
+    tdef = tooldefs_by_name.get(tool_name) if tool_name else None
+    description = tdef.description if tdef else None
     env = ResultEnvelope(
         tool_key=tool_name or "tool",
         name=_extract_operation_from_tool_name(tool_name or "tool"),
@@ -224,13 +223,15 @@ def _extract_operation_from_tool_name(full: str) -> str:
     return env.model_dump()
 
 
-async def build_adapters_for_tooldefs(
-    tool_manager: Any, tooldefs: List[Any]
-) -> tuple[list[dict], list[Any]]:
-    """Create OpenAI tool schemas and LangChain StructuredTool adapters for ToolDefinitions.
+async def build_adapters_for_tooldefs(tool_manager: Any, tooldefs: List[Any]) -> list[Any]:
+    """Create LangChain StructuredTool adapters for ToolDefinitions.
 
-    Returns (tool_schemas, adapters)
+    Each adapter wraps :meth:`ToolManager.resolve_tool_call` so that tools can
+    be executed either via LangGraph's :class:`ToolNode` or directly via the
+    manager. The same adapters can also be passed to ``ChatOpenAI.bind_tools``
+    so we do not need to maintain separate OpenAI-specific tool schemas.
     """
+
     try:
         from typing import Any as _Any
 
@@ -241,7 +242,7 @@ async def build_adapters_for_tooldefs(
         from pydantic import create_model as _create_model
     except Exception:
         # Best-effort fallback (should not happen in runtime)
-        return [], []
+        return []
 
     def _args_model_from_parameters(tool_name: str, params: dict) -> type[_BaseModel]:
         props = (params or {}).get("properties", {}) or {}
@@ -261,20 +262,19 @@ def _args_model_from_parameters(tool_name: str, params: dict) -> type[_BaseModel
             pass
         return args_model
 
-    tool_schemas: list[dict] = [t.to_openai_schema() for t in (tooldefs or [])]
     adapters: list[_StructuredTool] = []
     for tdef in tooldefs or []:
 
         async def _exec_fn(_name=tdef.name, **kwargs):
             return await tool_manager.resolve_tool_call(_name, kwargs or {})
 
-        ArgsModel = _args_model_from_parameters(tdef.name, getattr(tdef, "parameters", {}) or {})  # noqa: N806
+        args_model = _args_model_from_parameters(tdef.name, tdef.parameters or {})
         adapters.append(
             _StructuredTool.from_function(
                 coroutine=_exec_fn,
                 name=tdef.name,
-                description=getattr(tdef, "description", "") or "",
-                args_schema=ArgsModel,
+                description=tdef.description or "",
+                args_schema=args_model,
             )
         )
-    return tool_schemas, adapters
+    return adapters

redis_sre_agent/agent/knowledge_agent.py

@@ -97,18 +97,13 @@ def __init__(self, progress_callback: Optional[Callable[[str, str], Awaitable[No
 
         logger.info("Knowledge-only agent initialized (tools loaded per-query)")
 
-    def _build_workflow(self, tool_mgr: ToolManager) -> StateGraph:
+    def _build_workflow(self, tool_mgr: ToolManager, llm_with_tools: ChatOpenAI) -> StateGraph:
         """Build the LangGraph workflow for knowledge-only queries.
 
         Args:
             tool_mgr: ToolManager instance with knowledge tools loaded
         """
 
-        # Bind tools to LLM for this workflow
-        tools = tool_mgr.get_tools()
-        tool_schemas = [tool.to_openai_schema() for tool in tools]
-        llm_with_tools = self.llm.bind_tools(tool_schemas)
-
         async def agent_node(state: KnowledgeAgentState) -> KnowledgeAgentState:
             """Main agent node for knowledge queries."""
             messages = state["messages"]
@@ -131,25 +126,23 @@ def _sanitize_messages_for_llm(msgs: list[BaseMessage]) -> list[BaseMessage]:
                 for m in msgs:
                     if isinstance(m, AIMessage):
                         try:
-                            for tc in getattr(m, "tool_calls", []) or []:
+                            for tc in m.tool_calls or []:
                                 if isinstance(tc, dict):
                                     tid = tc.get("id") or tc.get("tool_call_id")
                                     if tid:
                                         seen_tool_ids.add(tid)
                         except Exception:
                             pass
                         clean.append(m)
-                    elif isinstance(m, _TM) or getattr(m, "type", "") == "tool":
-                        tid = getattr(m, "tool_call_id", None)
+                    elif isinstance(m, _TM) or m.type == "tool":
+                        tid = m.tool_call_id
                         if tid and tid in seen_tool_ids:
                             clean.append(m)
                         else:
                             continue
                     else:
                         clean.append(m)
-                while clean and (
-                    isinstance(clean[0], _TM) or getattr(clean[0], "type", "") == "tool"
-                ):
+                while clean and (isinstance(clean[0], _TM) or clean[0].type == "tool"):
                     clean = clean[1:]
                 return clean
 
@@ -192,8 +185,8 @@ def _sanitize_messages_for_llm(msgs: list[BaseMessage]) -> list[BaseMessage]:
                 )
                 # Coerce non-Message responses (e.g., simple mocks) into AIMessage
                 if not isinstance(response, BaseMessage):
-                    content = getattr(response, "content", None)
-                    tool_calls = getattr(response, "tool_calls", None)
+                    content = response.content
+                    tool_calls = response.tool_calls
                     response = AIMessage(
                         content=str(content) if content is not None else "", tool_calls=tool_calls
                     )
@@ -205,7 +198,7 @@ def _sanitize_messages_for_llm(msgs: list[BaseMessage]) -> list[BaseMessage]:
                 state["messages"] = messages + [response]
 
                 # Store tool calls for potential execution
-                if hasattr(response, "tool_calls") and response.tool_calls:
+                if response.tool_calls:
                     state["current_tool_calls"] = [
                         {
                             "id": tc.get("id", ""),
@@ -235,14 +228,14 @@ async def safe_tool_node(state: KnowledgeAgentState) -> KnowledgeAgentState:
             last_message = messages[-1] if messages else None
 
             # Verify we have tool calls to execute
-            if not (hasattr(last_message, "tool_calls") and last_message.tool_calls):
+            if not (last_message and last_message.tool_calls):
                 logger.warning("safe_tool_node called without tool_calls in last message")
                 return state
 
             try:
                 # Emit provider-supplied status updates before executing tools
                 try:
-                    pending = getattr(last_message, "tool_calls", []) or []
+                    pending = last_message.tool_calls or []
                     if self.progress_callback:
                         for tc in pending:
                             tool_name = tc.get("name")
@@ -255,9 +248,7 @@ async def safe_tool_node(state: KnowledgeAgentState) -> KnowledgeAgentState:
                     pass
 
                 # Execute tools using ToolManager (wrapped in OTel span)
-                _tool_names = [
-                    tc.get("name", "") for tc in (getattr(last_message, "tool_calls", []) or [])
-                ]
+                _tool_names = [tc.get("name", "") for tc in (last_message.tool_calls or [])]
                 with tracer.start_as_current_span(
                     "knowledge.tools.execute",
                     attributes={
@@ -347,7 +338,7 @@ def should_continue(state: KnowledgeAgentState) -> str:
             try:
                 from redis_sre_agent.core.config import settings as _settings
 
-                _budget = int(getattr(_settings, "max_tool_calls_per_stage", 3))
+                _budget = int(_settings.max_tool_calls_per_stage)
             except Exception:
                 _budget = 3
             prev_exec = int(state.get("tool_calls_executed", 0) or 0)
@@ -359,11 +350,7 @@ def should_continue(state: KnowledgeAgentState) -> str:
                 return END
 
             # If the last message has tool calls, execute them
-            if (
-                hasattr(last_message, "tool_calls")
-                and last_message.tool_calls
-                and len(last_message.tool_calls) > 0
-            ):
+            if last_message.tool_calls and len(last_message.tool_calls) > 0:
                 return "tools"
 
             return END
@@ -434,10 +421,17 @@ async def process_query(
 
         # Create ToolManager with Redis instance-independent tools
         async with ToolManager(redis_instance=None) as tool_mgr:
-            logger.info(f"Loaded {len(tool_mgr.get_tools())} usable without Redis instance details")
+            tools = tool_mgr.get_tools()
+            logger.info(f"Loaded {len(tools)} tools usable without Redis instance details")
+
+            # Build StructuredTool adapters and bind them to the LLM
+            from .helpers import build_adapters_for_tooldefs as _build_adapters
+
+            adapters = await _build_adapters(tool_mgr, tools)
+            llm_with_tools = self.llm.bind_tools(adapters)
 
-            # Build workflow with tools
-            workflow = self._build_workflow(tool_mgr)
+            # Build workflow with tools and bound LLM
+            workflow = self._build_workflow(tool_mgr, llm_with_tools)
 
             # Create initial state with conversation history
             initial_messages = []