style: fix linting and formatting

Author: LZH-YS1998

tools/code_implementation_server.py

@@ -430,8 +430,13 @@ async def read_code_mem(file_paths: List[str]) -> str:
  """
  try:
  if not file_paths or not isinstance(file_paths, list):
- result = {"status": "error", "message": "file_paths parameter is required and must be a list"}
- log_operation("read_code_mem_error", {"error": "missing_or_invalid_file_paths"})
+ result = {
+ "status": "error",
+ "message": "file_paths parameter is required and must be a list",
+ }
+ log_operation(
+ "read_code_mem_error", {"error": "missing_or_invalid_file_paths"}
+ )
  return json.dumps(result, ensure_ascii=False, indent=2)
 
  # Remove duplicates while preserving order
@@ -449,10 +454,11 @@ async def read_code_mem(file_paths: List[str]) -> str:
  "status": "no_summary",
  "file_paths": unique_file_paths,
  "message": "No summary file found.",
- "results": []
+ "results": [],
  }
  log_operation(
- "read_code_mem", {"file_paths": unique_file_paths, "status": "no_summary_file"}
+ "read_code_mem",
+ {"file_paths": unique_file_paths, "status": "no_summary_file"},
  )
  return json.dumps(result, ensure_ascii=False, indent=2)
 
@@ -465,37 +471,40 @@ async def read_code_mem(file_paths: List[str]) -> str:
  "status": "no_summary",
  "file_paths": unique_file_paths,
  "message": "Summary file is empty.",
- "results": []
+ "results": [],
  }
  log_operation(
- "read_code_mem", {"file_paths": unique_file_paths, "status": "empty_summary"}
+ "read_code_mem",
+ {"file_paths": unique_file_paths, "status": "empty_summary"},
  )
  return json.dumps(result, ensure_ascii=False, indent=2)
 
  # Process each file path and collect results
  results = []
  summaries_found = 0
- 
+
  for file_path in unique_file_paths:
  # Extract file-specific section from summary
- file_section = _extract_file_section_from_summary(summary_content, file_path)
- 
+ file_section = _extract_file_section_from_summary(
+ summary_content, file_path
+ )
+
  if file_section:
  file_result = {
  "file_path": file_path,
  "status": "summary_found",
  "summary_content": file_section,
- "message": f"Summary information found for {file_path}"
+ "message": f"Summary information found for {file_path}",
  }
  summaries_found += 1
  else:
  file_result = {
  "file_path": file_path,
  "status": "no_summary",
  "summary_content": None,
- "message": f"No summary found for {file_path}"
+ "message": f"No summary found for {file_path}",
  }
- 
+
  results.append(file_result)
 
  # Determine overall status
@@ -512,7 +521,7 @@ async def read_code_mem(file_paths: List[str]) -> str:
  "total_requested": len(unique_file_paths),
  "summaries_found": summaries_found,
  "message": f"Found summaries for {summaries_found}/{len(unique_file_paths)} files",
- "results": results
+ "results": results,
  }
 
  log_operation(
@@ -531,10 +540,14 @@ async def read_code_mem(file_paths: List[str]) -> str:
  result = {
  "status": "error",
  "message": f"Failed to check code memory: {str(e)}",
- "file_paths": file_paths if isinstance(file_paths, list) else [str(file_paths)],
- "results": []
+ "file_paths": file_paths
+ if isinstance(file_paths, list)
+ else [str(file_paths)],
+ "results": [],
  }
- log_operation("read_code_mem_error", {"file_paths": file_paths, "error": str(e)})
+ log_operation(
+ "read_code_mem_error", {"file_paths": file_paths, "error": str(e)}
+ )
  return json.dumps(result, ensure_ascii=False, indent=2)
 
 

utils/llm_utils.py

@@ -53,38 +53,37 @@ def get_preferred_llm_class(config_path: str = "mcp_agent.secrets.yaml") -> Type
  return OpenAIAugmentedLLM
 
 
-def get_default_models(config_path: str = "mcp_agent.config.yaml") -> dict:
+def get_default_models(config_path: str = "mcp_agent.config.yaml"):
  """
- Get default models configuration from config file.
+ Get default models from configuration file.
 
  Args:
- config_path: Path to the main configuration file
+ config_path: Path to the configuration file
 
  Returns:
- dict: Default models configuration
+ dict: Dictionary with 'anthropic' and 'openai' default models
  """
  try:
  if os.path.exists(config_path):
  with open(config_path, "r", encoding="utf-8") as f:
  config = yaml.safe_load(f)
 
- # Extract model configurations
- openai_config = config.get("openai", {})
- anthropic_config = config.get("anthropic", {})
+ # Handle null values in config sections
+ anthropic_config = config.get("anthropic") or {}
+ openai_config = config.get("openai") or {}
 
- return {
- "anthropic": anthropic_config.get(
-  "default_model", "claude-sonnet-4-20250514"
-  ),
- "openai": openai_config.get("default_model", "o3-mini"),
- }
+ anthropic_model = anthropic_config.get(
+ "default_model", "claude-sonnet-4-20250514"
+ )
+ openai_model = openai_config.get("default_model", "o3-mini")
+
+ return {"anthropic": anthropic_model, "openai": openai_model}
  else:
- print(f"🤖 Config file {config_path} not found, using default models")
+ print(f"Config file {config_path} not found, using default models")
  return {"anthropic": "claude-sonnet-4-20250514", "openai": "o3-mini"}
 
  except Exception as e:
- print(f"🤖 Error reading config file {config_path}: {e}")
- print("🤖 Falling back to default models")
+ print(f"❌Error reading config file {config_path}: {e}")
  return {"anthropic": "claude-sonnet-4-20250514", "openai": "o3-mini"}
 
 

workflows/agents/code_implementation_agent.py

@@ -310,7 +310,8 @@ async def _handle_read_file_with_memory_optimization(self, tool_call: Dict) -> D
  result_data = json.loads(read_code_mem_result)
  # Check if any summaries were found in the results
  should_use_summary = (
- result_data.get("status") in ["all_summaries_found", "partial_summaries_found"]
+ result_data.get("status")
+ in ["all_summaries_found", "partial_summaries_found"]
  and result_data.get("summaries_found", 0) > 0
  )
  except json.JSONDecodeError:
@@ -339,17 +340,25 @@ async def _handle_read_file_with_memory_optimization(self, tool_call: Dict) -> D
  # Extract the specific file result for the single file we requested
  file_results = result_data.get("results", [])
  if file_results and len(file_results) > 0:
- specific_result = file_results[0] # Get the first (and only) result
+ specific_result = file_results[
+ 0
+ ] # Get the first (and only) result
  # Transform to match the old single-file format for backward compatibility
  transformed_result = {
  "status": specific_result.get("status", "no_summary"),
- "file_path": specific_result.get("file_path", file_path),
- "summary_content": specific_result.get("summary_content"),
+ "file_path": specific_result.get(
+ "file_path", file_path
+ ),
+ "summary_content": specific_result.get(
+ "summary_content"
+ ),
  "message": specific_result.get("message", ""),
  "original_tool": "read_file",
- "optimization": "redirected_to_read_code_mem"
+ "optimization": "redirected_to_read_code_mem",
  }
- final_result = json.dumps(transformed_result, ensure_ascii=False)
+ final_result = json.dumps(
+ transformed_result, ensure_ascii=False
+ )
  else:
  # Fallback if no results
  result_data["original_tool"] = "read_file"
@@ -942,7 +951,11 @@ async def test_summary_functionality(self, test_file_path: str = None):
  json.loads(result) if isinstance(result, str) else result
  )
 
- if result_data.get("status") in ["all_summaries_found", "partial_summaries_found"] and result_data.get("summaries_found", 0) > 0:
+ if (
+ result_data.get("status")
+ in ["all_summaries_found", "partial_summaries_found"]
+ and result_data.get("summaries_found", 0) > 0
+ ):
  summary_files_found += 1
  except Exception as e:
  self.logger.warning(
@@ -1042,7 +1055,11 @@ async def test_summary_optimization(self, test_file_path: str = "config.py"):
 
  result_data = json.loads(result) if isinstance(result, str) else result
 
- return result_data.get("status") in ["all_summaries_found", "partial_summaries_found"] and result_data.get("summaries_found", 0) > 0
+ return (
+ result_data.get("status")
+ in ["all_summaries_found", "partial_summaries_found"]
+ and result_data.get("summaries_found", 0) > 0
+ )
  except Exception as e:
  self.logger.warning(f"Failed to test read_code_mem optimization: {e}")
  return False