Add progress notifications to ChatAgent tool execution and cleanup

- Add emitter parameter to ChatAgent._build_workflow() to enable progress notifications
- Emit tool_call updates in ChatAgent.tool_node() before executing tools
- Remove duplicate imports in docket_tasks.py (process_chat_turn, process_knowledge_query)
- Remove unused API_BASE_URL and os import from mcp_server/server.py
- Fix ASGI app initialization in mcp_server/server.py for uvicorn
- Remove unused ThreadManager import from progress.py
- Fix import ordering and remove unused imports across test files
- Apply ruff format to all modified files

Author: abrookins

redis_sre_agent/agent/chat_agent.py

@@ -260,7 +260,9 @@ async def agent_node(state: ChatAgentState) -> Dict[str, Any]:
             return {
                 "messages": new_messages,
                 "iteration_count": iteration_count + 1,
-                "current_tool_calls": response.tool_calls if hasattr(response, "tool_calls") else [],
+                "current_tool_calls": response.tool_calls
+                if hasattr(response, "tool_calls")
+                else [],
             }
 
         async def tool_node(state: ChatAgentState) -> Dict[str, Any]:
@@ -277,7 +279,9 @@ async def tool_node(state: ChatAgentState) -> Dict[str, Any]:
             # Emit progress updates for each tool call
             if emitter and tool_calls:
                 for tc in tool_calls:
-                    tool_name = tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+                    tool_name = (
+                        tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+                    )
                     tool_args = (
                         tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", {})
                     ) or {}
@@ -299,7 +303,9 @@ async def tool_node(state: ChatAgentState) -> Dict[str, Any]:
 
                 # Build envelopes for each tool call result
                 for idx, tc in enumerate(tool_calls):
-                    tool_name = tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+                    tool_name = (
+                        tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None)
+                    )
                     tool_args = (
                         tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", {})
                     ) or {}
@@ -445,9 +451,7 @@ async def process_query(
             thread_config = {"configurable": {"thread_id": session_id}}
 
             try:
-                await emitter.emit(
-                    "Chat agent processing your question...", "agent_start"
-                )
+                await emitter.emit("Chat agent processing your question...", "agent_start")
 
                 final_state = await app.ainvoke(initial_state, config=thread_config)
 

redis_sre_agent/agent/knowledge_agent.py

@@ -510,9 +510,7 @@ async def process_query(
                 logger.error(f"Knowledge agent processing failed: {e}")
                 error_response = f"I encountered an error while processing your knowledge query: {str(e)}. Please try asking a more specific question about SRE practices, troubleshooting methodologies, or system reliability concepts."
 
-                await emitter.emit(
-                    f"Knowledge agent encountered an error: {str(e)}", "agent_error"
-                )
+                await emitter.emit(f"Knowledge agent encountered an error: {str(e)}", "agent_error")
 
                 return error_response
 

redis_sre_agent/agent/langgraph_agent.py

@@ -598,8 +598,7 @@ async def _summarize_envelopes_for_reasoning(
                 batch_prompt += "\n\n"
 
             batch_prompt += (
-                "Return JSON array format: "
-                '[{"summary": "key findings..."}, {"summary": "..."}]'
+                'Return JSON array format: [{"summary": "key findings..."}, {"summary": "..."}]'
             )
 
             try:
@@ -623,9 +622,7 @@ async def _summarize_envelopes_for_reasoning(
                     pass
 
                 # Apply summaries to envelopes
-                for j, (orig_idx, env) in enumerate(
-                    zip(to_summarize_indices, to_summarize)
-                ):
+                for j, (orig_idx, env) in enumerate(zip(to_summarize_indices, to_summarize)):
                     summary_text = (
                         summaries[j].get("summary", "")
                         if j < len(summaries) and isinstance(summaries[j], dict)
@@ -1236,9 +1233,7 @@ def _sev_score(t: dict) -> int:
                     )
                     # Use summarized envelopes for recommendation workers
                     # LLM can call expand_evidence to get full details if needed
-                    env_by_key = {
-                        e.get("tool_key"): e for e in summarized_envelopes
-                    }
+                    env_by_key = {e.get("tool_key"): e for e in summarized_envelopes}
                     for t in topics:
                         ev_keys = [k for k in (t.get("evidence_keys") or []) if isinstance(k, str)]
                         ev = [env_by_key[k] for k in ev_keys if k in env_by_key]

redis_sre_agent/cli/query.py

@@ -150,7 +150,9 @@ async def _query():
 
             # Show thread ID for follow-up queries
             console.print("\n[dim]💡 To continue this conversation:[/dim]")
-            console.print(f"[dim]   redis-sre-agent query --thread-id {active_thread_id} \"your follow-up\"[/dim]")
+            console.print(
+                f'[dim]   redis-sre-agent query --thread-id {active_thread_id} "your follow-up"[/dim]'
+            )
 
         except Exception as e:
             console.print(f"[red]❌ Error: {e}[/red]")

redis_sre_agent/core/docket_tasks.py

@@ -220,7 +220,13 @@ async def process_chat_turn(
         # Add response to thread as assistant message
         await thread_manager.append_messages(
             thread_id,
-            [{"role": "assistant", "content": response, "metadata": {"task_id": task_id, "agent": "chat"}}],
+            [
+                {
+                    "role": "assistant",
+                    "content": response,
+                    "metadata": {"task_id": task_id, "agent": "chat"},
+                }
+            ],
         )
 
         return result
@@ -290,7 +296,13 @@ async def process_knowledge_query(
         # Add response to thread as assistant message
         await thread_manager.append_messages(
             thread_id,
-            [{"role": "assistant", "content": response, "metadata": {"task_id": task_id, "agent": "knowledge"}}],
+            [
+                {
+                    "role": "assistant",
+                    "content": response,
+                    "metadata": {"task_id": task_id, "agent": "knowledge"},
+                }
+            ],
         )
 
         return result
@@ -653,7 +665,9 @@ async def process_agent_turn(
             agent = get_sre_agent()
         elif agent_type == AgentType.REDIS_CHAT:
             # Get the target instance for the chat agent
-            target_instance = await get_instance_by_id(active_instance_id) if active_instance_id else None
+            target_instance = (
+                await get_instance_by_id(active_instance_id) if active_instance_id else None
+            )
             agent = get_chat_agent(redis_instance=target_instance)
         else:
             agent = get_knowledge_agent()

redis_sre_agent/core/progress.py

@@ -222,9 +222,7 @@ async def emit(
     ) -> None:
         """Emit notification to task storage."""
         try:
-            await self._task_manager.add_task_update(
-                self._task_id, message, update_type, metadata
-            )
+            await self._task_manager.add_task_update(self._task_id, message, update_type, metadata)
         except Exception as e:
             # Best-effort: don't fail the agent if notification logging fails
             logger.warning(f"Failed to emit task notification: {e}")

redis_sre_agent/core/redis.py

@@ -235,9 +235,7 @@ def get_vectorizer() -> OpenAITextVectorizer:
         redis_url=redis_url,
         ttl=settings.embeddings_cache_ttl,
     )
-    logger.debug(
-        f"Vectorizer created with embeddings cache (ttl={settings.embeddings_cache_ttl}s)"
-    )
+    logger.debug(f"Vectorizer created with embeddings cache (ttl={settings.embeddings_cache_ttl}s)")
 
     return OpenAITextVectorizer(
         model=settings.embedding_model,

redis_sre_agent/tools/manager.py

@@ -239,7 +239,9 @@ async def _load_mcp_providers(self) -> None:
         # Build set of excluded capabilities for fast lookup
         excluded_caps = set(self.exclude_mcp_categories or [])
         if excluded_caps:
-            logger.info(f"MCP tools with these categories will be excluded: {[c.value for c in excluded_caps]}")
+            logger.info(
+                f"MCP tools with these categories will be excluded: {[c.value for c in excluded_caps]}"
+            )
 
         for server_name, server_config in settings.mcp_servers.items():
             try:

tests/unit/agent/test_chat_agent.py

@@ -12,7 +12,6 @@
 from redis_sre_agent.core.progress import (
     CallbackEmitter,
     NullEmitter,
-    ProgressEmitter,
 )
 
 
@@ -121,6 +120,7 @@ def test_get_chat_agent_without_instance(self):
 
             # Clear cache
             from redis_sre_agent.agent import chat_agent
+
             chat_agent._chat_agents.clear()
 
             agent = get_chat_agent()
@@ -137,6 +137,7 @@ def test_get_chat_agent_caches_by_instance_name(self):
 
             # Clear cache
             from redis_sre_agent.agent import chat_agent
+
             chat_agent._chat_agents.clear()
 
             instance1 = RedisInstance(

tests/unit/agent/test_envelope_summarization.py

@@ -1,8 +1,9 @@
 """Tests for envelope summarization and expand_evidence tool in the reasoning phase."""
 
-import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import pytest
+
 from redis_sre_agent.agent.langgraph_agent import SRELangGraphAgent
 
 
@@ -95,9 +96,7 @@ async def test_mixed_envelopes_partial_summarization(self, agent):
         mock_response.content = '[{"summary": "Large content summarized"}]'
         agent.mini_llm.ainvoke = AsyncMock(return_value=mock_response)
 
-        result = await agent._summarize_envelopes_for_reasoning(
-            [small_envelope, large_envelope]
-        )
+        result = await agent._summarize_envelopes_for_reasoning([small_envelope, large_envelope])
 
         assert len(result) == 2
         # Small envelope unchanged
@@ -109,8 +108,13 @@ async def test_mixed_envelopes_partial_summarization(self, agent):
     async def test_order_preserved(self, agent):
         """Test that envelope order is preserved after summarization."""
         envelopes = [
-            {"tool_key": f"tool_{i}", "name": f"t{i}", "args": {}, "status": "success",
-             "data": {"id": i, "content": "x" * (100 if i % 2 == 0 else 1000)}}
+            {
+                "tool_key": f"tool_{i}",
+                "name": f"t{i}",
+                "args": {},
+                "status": "success",
+                "data": {"id": i, "content": "x" * (100 if i % 2 == 0 else 1000)},
+            }
             for i in range(5)
         ]