diff --git a/.env.example b/.env.example index a7e6376..6435962 100644 --- a/.env.example +++ b/.env.example @@ -108,3 +108,9 @@ MAX_CONVERSATION_TURNS=20 # ERROR: Shows only errors LOG_LEVEL=DEBUG +# Optional: Language/Locale for AI responses +# When set, all AI tools will respond in the specified language +# while maintaining their analytical capabilities +# Examples: "fr-FR", "en-US", "zh-CN", "zh-TW", "ja-JP", "ko-KR", "es-ES" +# Leave empty for default language (English) +# LOCALE=fr-FR diff --git a/config.py b/config.py index bd330eb..fb9663d 100644 --- a/config.py +++ b/config.py @@ -136,6 +136,15 @@ def _calculate_mcp_prompt_limit() -> int: MCP_PROMPT_SIZE_LIMIT = _calculate_mcp_prompt_limit() +# Language/Locale Configuration +# LOCALE: Language/locale specification for AI responses +# When set, all AI tools will respond in the specified language while +# maintaining their analytical capabilities +# Examples: "fr-FR", "en-US", "zh-CN", "zh-TW", "ja-JP", "ko-KR", "es-ES", +# "de-DE", "it-IT", "pt-PT" +# Leave empty for default language (English) +LOCALE = os.getenv("LOCALE", "") + # Threading configuration # Simple in-memory conversation threading for stateless MCP environment # Conversations persist only during the Claude session diff --git a/docs/locale-configuration.md b/docs/locale-configuration.md new file mode 100644 index 0000000..ffac4ef --- /dev/null +++ b/docs/locale-configuration.md @@ -0,0 +1,186 @@ +# Locale Configuration for Zen MCP Server + +This guide explains how to configure and use the localization feature to customize the language of responses from MCP tools. + +## Overview + +The localization feature allows you to specify the language in which MCP tools should respond, while maintaining their analytical capabilities. This is especially useful for non-English speakers who want to receive answers in their native language. + +## Configuration + +### 1. Environment Variable + +Set the language using the `LOCALE` environment variable in your `.env` file: + +```bash +# In your .env file +LOCALE=fr-FR +``` + +### 2. Supported Languages + +You can use any standard language code. Examples: + +- `fr-FR` - French (France) +- `en-US` - English (United States) +- `zh-CN` - Chinese (Simplified) +- `zh-TW` - Chinese (Traditional) +- `ja-JP` - Japanese +- `ko-KR` - Korean +- `es-ES` - Spanish (Spain) +- `de-DE` - German (Germany) +- `it-IT` - Italian (Italy) +- `pt-PT` - Portuguese (Portugal) +- `ru-RU` - Russian (Russia) +- `ar-SA` - Arabic (Saudi Arabia) + +### 3. Default Behavior + +If no language is specified (`LOCALE` is empty or unset), tools will default to English. + +## Technical Implementation + +### Architecture + +Localization is implemented in the `BaseTool` class in `tools/shared/base_tool.py`. All tools inherit this feature automatically. + +### `get_language_instruction()` Method + +```python +def get_language_instruction(self) -> str: + """ + Generate language instruction based on LOCALE configuration. + Returns: + str: Language instruction to prepend to prompt, or empty string if no locale set + """ + from config import LOCALE + if not LOCALE or not LOCALE.strip(): + return "" + return f"Always respond in {LOCALE.strip()}.\n\n" +``` + +### Integration in Tool Execution + +The language instruction is automatically prepended to the system prompt of each tool: + +```python +# In tools/simple/base.py +base_system_prompt = self.get_system_prompt() +language_instruction = self.get_language_instruction() +system_prompt = language_instruction + base_system_prompt +``` + +## Usage + +### 1. Basic Setup + +1. Edit your `.env` file: + ```bash + LOCALE=fr-FR + ``` +2. Restart the MCP server: + ```bash + python server.py + ``` +3. Use any tool – responses will be in the specified language. + +### 2. Example + +**Before (default English):** +``` +Tool: chat +Input: "Explain how to use Python dictionaries" +Output: "Python dictionaries are key-value pairs that allow you to store and organize data..." +``` + +**After (with LOCALE=fr-FR):** +``` +Tool: chat +Input: "Explain how to use Python dictionaries" +Output: "Les dictionnaires Python sont des paires clé-valeur qui permettent de stocker et d'organiser des données..." +``` + +### 3. Affected Tools + +All MCP tools are affected by this configuration: + +- `chat` – General conversation +- `codereview` – Code review +- `analyze` – Code analysis +- `debug` – Debugging +- `refactor` – Refactoring +- `thinkdeep` – Deep thinking +- `consensus` – Model consensus +- And all other tools... + +## Best Practices + +### 1. Language Choice +- Use standard language codes (ISO 639-1 with ISO 3166-1 country codes) +- Be specific with regional variants if needed (e.g., `zh-CN` vs `zh-TW`) + +### 2. Consistency +- Use the same language setting across your team for consistency +- Document the chosen language in your team documentation + +### 3. Testing +- Test the configuration with different tools to ensure consistency + +## Troubleshooting + +### Issue: Language does not change +**Solution:** +1. Check that the `LOCALE` variable is correctly set in `.env` +2. Fully restart the MCP server +3. Ensure there are no extra spaces in the value + +### Issue: Partially translated responses +**Explanation:** +- AI models may sometimes mix languages +- This depends on the multilingual capabilities of the model used +- Technical terms may remain in English + +### Issue: Configuration errors +**Solution:** +1. Check the syntax of your `.env` file +2. Make sure there are no quotes around the value + +## Advanced Customization + +### Customizing the Language Instruction + +To customize the language instruction, modify the `get_language_instruction()` method in `tools/shared/base_tool.py`: + +```python +def get_language_instruction(self) -> str: + from config import LOCALE + if not LOCALE or not LOCALE.strip(): + return "" + # Custom instruction + return f"Always respond in {LOCALE.strip()} and use a professional tone.\n\n" +``` + +### Per-Tool Customization + +You can also override the method in specific tools for custom behavior: + +```python +class MyCustomTool(SimpleTool): + def get_language_instruction(self) -> str: + from config import LOCALE + if LOCALE == "fr-FR": + return "Respond in French with precise technical vocabulary.\n\n" + elif LOCALE == "zh-CN": + return "请用中文回答，使用专业术语。\n\n" + else: + return super().get_language_instruction() +``` + +## Integration with Other Features + +Localization works with all other MCP server features: + +- **Conversation threading** – Multilingual conversations are supported +- **File processing** – File analysis is in the specified language +- **Web search** – Search instructions remain functional +- **Model selection** – Works with all supported models diff --git a/providers/openai_compatible.py b/providers/openai_compatible.py index fec4484..9a7846b 100644 --- a/providers/openai_compatible.py +++ b/providers/openai_compatible.py @@ -311,11 +311,10 @@ class OpenAICompatibleProvider(ModelProvider): last_exception = None for attempt in range(max_retries): - try: - # Log the exact payload being sent for debugging + try: # Log the exact payload being sent for debugging import json - logging.info(f"o3-pro API request payload: {json.dumps(completion_params, indent=2)}") + logging.info(f"o3-pro API request payload: {json.dumps(completion_params, indent=2, ensure_ascii=False)}") # Use OpenAI client's responses endpoint response = self.client.responses.create(**completion_params) diff --git a/simulator_tests/base_test.py b/simulator_tests/base_test.py index ec1a95e..cbe41b9 100644 --- a/simulator_tests/base_test.py +++ b/simulator_tests/base_test.py @@ -136,10 +136,12 @@ class Calculator: "id": 2, "method": "tools/call", "params": {"name": tool_name, "arguments": params}, - } - - # Combine all messages - messages = [json.dumps(init_request), json.dumps(initialized_notification), json.dumps(tool_request)] + } # Combine all messages + messages = [ + json.dumps(init_request, ensure_ascii=False), + json.dumps(initialized_notification, ensure_ascii=False), + json.dumps(tool_request, ensure_ascii=False) + ] # Join with newlines as MCP expects input_data = "\n".join(messages) + "\n" diff --git a/simulator_tests/test_analyze_validation.py b/simulator_tests/test_analyze_validation.py index dd431ca..e9d1160 100644 --- a/simulator_tests/test_analyze_validation.py +++ b/simulator_tests/test_analyze_validation.py @@ -112,11 +112,9 @@ class UserService: result = await self.db.execute( "SELECT * FROM users WHERE id = %s", (user_id,) ) - user_data = result.fetchone() - - if user_data: + user_data = result.fetchone() if user_data: # Cache for 1 hour - magic number - self.cache.setex(cache_key, 3600, json.dumps(user_data)) + self.cache.setex(cache_key, 3600, json.dumps(user_data, ensure_ascii=False)) return user_data @@ -273,10 +271,8 @@ class UserProfile(Base): try: return json.loads(self.preferences) if self.preferences else {} except json.JSONDecodeError: - return {} - - def set_preferences(self, prefs: dict): - self.preferences = json.dumps(prefs) + return {} def set_preferences(self, prefs: dict): + self.preferences = json.dumps(prefs, ensure_ascii=False) class AuditLog(Base): __tablename__ = "audit_logs" @@ -298,7 +294,7 @@ class AuditLog(Base): log = cls( user_id=user_id, action=action, - details=json.dumps(details) if details else None, + details=json.dumps(details, ensure_ascii=False) if details else None, ip_address=ip_address, user_agent=user_agent ) @@ -692,9 +688,7 @@ class PerformanceTimer: if not response_final_data.get("analysis_complete"): self.logger.error("Expected analysis_complete=true for final step") - return False - - # Check for expert analysis + return False # Check for expert analysis if "expert_analysis" not in response_final_data: self.logger.error("Missing expert_analysis in final response") return False @@ -702,7 +696,7 @@ class PerformanceTimer: expert_analysis = response_final_data.get("expert_analysis", {}) # Check for expected analysis content (checking common patterns) - analysis_text = json.dumps(expert_analysis).lower() + analysis_text = json.dumps(expert_analysis, ensure_ascii=False).lower() # Look for architectural analysis indicators arch_indicators = ["architecture", "pattern", "coupling", "dependency", "scalability", "maintainability"] diff --git a/simulator_tests/test_codereview_validation.py b/simulator_tests/test_codereview_validation.py index 9aac59d..2bac993 100644 --- a/simulator_tests/test_codereview_validation.py +++ b/simulator_tests/test_codereview_validation.py @@ -514,7 +514,7 @@ class ConfigurationManager: expert_analysis = response_final_data.get("expert_analysis", {}) # Check for expected analysis content (checking common patterns) - analysis_text = json.dumps(expert_analysis).lower() + analysis_text = json.dumps(expert_analysis, ensure_ascii=False).lower() # Look for code review identification review_indicators = ["security", "vulnerability", "performance", "critical", "api", "key"] diff --git a/simulator_tests/test_debug_validation.py b/simulator_tests/test_debug_validation.py index a5933e1..eb1de81 100644 --- a/simulator_tests/test_debug_validation.py +++ b/simulator_tests/test_debug_validation.py @@ -385,7 +385,7 @@ RuntimeError: dictionary changed size during iteration expert_analysis = response_final_data.get("expert_analysis", {}) # Check for expected analysis content (checking common patterns) - analysis_text = json.dumps(expert_analysis).lower() + analysis_text = json.dumps(expert_analysis, ensure_ascii=False).lower() # Look for bug identification bug_indicators = ["dictionary", "iteration", "modify", "runtime", "error", "del"] diff --git a/simulator_tests/test_precommitworkflow_validation.py b/simulator_tests/test_precommitworkflow_validation.py index 851b047..1fefa77 100644 --- a/simulator_tests/test_precommitworkflow_validation.py +++ b/simulator_tests/test_precommitworkflow_validation.py @@ -430,7 +430,7 @@ REQUIREMENTS: expert_analysis = response_final_data.get("expert_analysis", {}) # Check for expected analysis content (checking common patterns) - analysis_text = json.dumps(expert_analysis).lower() + analysis_text = json.dumps(expert_analysis, ensure_ascii=False).lower() # Look for security issue identification security_indicators = ["sql", "injection", "security", "hardcoded", "secret", "authentication"] diff --git a/simulator_tests/test_refactor_validation.py b/simulator_tests/test_refactor_validation.py index 76940c9..d72b183 100644 --- a/simulator_tests/test_refactor_validation.py +++ b/simulator_tests/test_refactor_validation.py @@ -125,7 +125,7 @@ class DataProcessorManager: # Code smell: Duplicate date formatting logic if output_format == 'json': processed_data['processed_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') - result = json.dumps(processed_data) + result = json.dumps(processed_data, ensure_ascii=False) elif output_format == 'csv': processed_data['processed_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') result = f"{processed_data['full_name']},{processed_data['email_domain']},{processed_data['age_category']}" @@ -580,7 +580,7 @@ class UserData: self.logger.error("Missing expert_analysis in final response") return False expert_analysis = response_final_data.get("expert_analysis", {}) - analysis_content = json.dumps(expert_analysis).lower() + analysis_content = json.dumps(expert_analysis, ensure_ascii=False).lower() elif actual_status == "files_required_to_continue": # For files_required_to_continue, analysis is in content field if "content" not in response_final_data: @@ -708,7 +708,7 @@ def format_output(data, format_type): \"\"\"Format output - duplicate logic\"\"\" if format_type == 'json': import json - return json.dumps(data) + return json.dumps(data, ensure_ascii=False) elif format_type == 'csv': return ','.join(str(v) for v in data.values()) else: diff --git a/simulator_tests/test_testgen_validation.py b/simulator_tests/test_testgen_validation.py index 549140c..dfd1636 100644 --- a/simulator_tests/test_testgen_validation.py +++ b/simulator_tests/test_testgen_validation.py @@ -346,7 +346,7 @@ class TestCalculatorBasic: expert_analysis = response_final_data.get("expert_analysis", {}) # Check for expected analysis content - analysis_text = json.dumps(expert_analysis).lower() + analysis_text = json.dumps(expert_analysis, ensure_ascii=False).lower() # Look for test generation indicators test_indicators = ["test", "edge", "boundary", "error", "coverage", "pytest"] diff --git a/simulator_tests/test_thinkdeep_validation.py b/simulator_tests/test_thinkdeep_validation.py index f25b93f..ed6a0d1 100644 --- a/simulator_tests/test_thinkdeep_validation.py +++ b/simulator_tests/test_thinkdeep_validation.py @@ -415,7 +415,7 @@ class ThinkDeepWorkflowValidationTest(ConversationBaseTest): expert_analysis = {"analysis": expert_analysis} # Check for expected analysis content (checking common patterns) - analysis_text = json.dumps(expert_analysis).lower() + analysis_text = json.dumps(expert_analysis, ensure_ascii=False).lower() # Look for thinking analysis validation thinking_indicators = ["migration", "strategy", "microservices", "risk", "approach", "implementation"] diff --git a/tests/test_collaboration.py b/tests/test_collaboration.py index 431c89e..dbc0c9c 100644 --- a/tests/test_collaboration.py +++ b/tests/test_collaboration.py @@ -34,7 +34,8 @@ class TestDynamicContextRequests: "status": "files_required_to_continue", "mandatory_instructions": "I need to see the package.json file to understand dependencies", "files_needed": ["package.json", "package-lock.json"], - } + }, + ensure_ascii=False ) mock_provider = create_mock_provider() @@ -174,7 +175,8 @@ class TestDynamicContextRequests: ], }, }, - } + }, + ensure_ascii=False ) mock_provider = create_mock_provider() @@ -339,7 +341,8 @@ class TestCollaborationWorkflow: "status": "files_required_to_continue", "mandatory_instructions": "I need to see the package.json file to analyze npm dependencies", "files_needed": ["package.json", "package-lock.json"], - } + }, + ensure_ascii=False ) mock_provider = create_mock_provider() @@ -405,7 +408,8 @@ class TestCollaborationWorkflow: "status": "files_required_to_continue", "mandatory_instructions": "I need to see the configuration file to understand the connection settings", "files_needed": ["config.py"], - } + }, + ensure_ascii=False ) mock_provider = create_mock_provider() diff --git a/tests/test_integration_utf8.py b/tests/test_integration_utf8.py new file mode 100644 index 0000000..d6c28cd --- /dev/null +++ b/tests/test_integration_utf8.py @@ -0,0 +1,477 @@ +""" +Full integration test script to validate UTF-8 implementation +and French localization. + +This script runs all unit tests and checks full integration. +""" + +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + + +def run_utf8_integration_tests(): + """Run UTF-8 integration tests.""" + print("🚀 Starting UTF-8 integration tests") + print("=" * 60) + + # Test environment setup + os.environ["LOCALE"] = "fr-FR" + os.environ["GEMINI_API_KEY"] = "dummy-key-for-tests" + os.environ["OPENAI_API_KEY"] = "dummy-key-for-tests" + + # Test 1: Validate UTF-8 characters in json.dumps + print("\n1️⃣ UTF-8 encoding test with json.dumps") + test_utf8_json_encoding() + + # Test 2: Validate language instruction generation + print("\n2️⃣ Language instruction generation test") + test_language_instruction_generation() + + # Test 3: Validate UTF-8 file handling + print("\n3️⃣ UTF-8 file handling test") + test_file_utf8_handling() + + # Test 4: Validate MCP tools integration + print("\n4️⃣ MCP tools integration test") + test_mcp_tools_integration() + + # Test 5: Run unit tests + print("\n5️⃣ Running unit tests") + run_unit_tests() + + print("\n✅ All UTF-8 integration tests completed!") + print("🇫🇷 French localization works correctly!") + + +def test_utf8_json_encoding(): + """Test UTF-8 encoding with json.dumps(ensure_ascii=False).""" + print(" Testing UTF-8 JSON encoding...") + + # Test data with French characters and emojis + test_data = { + "analyse": { + "statut": "terminée", + "résultat": "Aucun problème critique détecté", + "recommandations": [ + "Améliorer la documentation", + "Optimiser les performances", + "Ajouter des tests unitaires", + ], + "métadonnées": { + "créé_par": "Développeur Principal", + "date_création": "2024-01-01", + "dernière_modification": "2024-01-15", + }, + "émojis_status": { + "critique": "🔴", + "élevé": "🟠", + "moyen": "🟡", + "faible": "🟢", + "succès": "✅", + "erreur": "❌", + }, + }, + "outils": [ + {"nom": "analyse", "description": "Analyse architecturale avancée"}, + {"nom": "révision", "description": "Révision de code automatisée"}, + {"nom": "génération", "description": "Génération de documentation"}, + ], + } + + # Test with ensure_ascii=False + json_correct = json.dumps(test_data, ensure_ascii=False, indent=2) + + # Checks + utf8_terms = [ + "terminée", + "résultat", + "détecté", + "Améliorer", + "créé_par", + "Développeur", + "création", + "métadonnées", + "dernière", + "émojis_status", + "élevé", + "révision", + "génération", + ] + + emojis = ["🔴", "🟠", "🟡", "🟢", "✅", "❌"] + + for term in utf8_terms: + assert term in json_correct, f"Missing UTF-8 term: {term}" + + for emoji in emojis: + assert emoji in json_correct, f"Missing emoji: {emoji}" + + # Check for escaped characters + assert "\\u" not in json_correct, "Escaped Unicode characters detected!" + + # Test parsing + parsed = json.loads(json_correct) + assert parsed["analyse"]["statut"] == "terminée" + assert parsed["analyse"]["émojis_status"]["critique"] == "🔴" + + print(" ✅ UTF-8 JSON encoding: SUCCESS") + + +def test_language_instruction_generation(): + """Test language instruction generation.""" + print(" Testing language instruction generation...") + + # Simulation of get_language_instruction + def get_language_instruction(): + locale = os.getenv("LOCALE", "").strip() + if not locale: + return "" + return f"Always respond in {locale}.\n\n" + + # Test with different locales + test_locales = [ + ("fr-FR", "French"), + ("en-US", "English"), + ("es-ES", "Spanish"), + ("de-DE", "German"), + ("", "none"), + ] + + for locale, description in test_locales: + os.environ["LOCALE"] = locale + instruction = get_language_instruction() + + if locale: + assert locale in instruction, f"Missing {locale} in instruction" + assert instruction.endswith("\n\n"), "Incorrect instruction format" + print(f" 📍 {description}: {instruction.strip()}") + else: + assert instruction == "", "Empty instruction expected for empty locale" + print(f" 📍 {description}: (empty)") + + # Restore French locale + os.environ["LOCALE"] = "fr-FR" + print(" ✅ Language instruction generation: SUCCESS") + + +def test_file_utf8_handling(): + """Test handling of files with UTF-8 content.""" + print(" Testing UTF-8 file handling...") + + # File content with French characters + french_content = '''#!/usr/bin/env python3 +""" +Module de gestion des préférences utilisateur. +Développé par: Équipe Technique +Date de création: 15 décembre 2024 +""" + +import json +from typing import Dict, Optional + +class GestionnairePreferences: + """Gestionnaire des préférences utilisateur avec support UTF-8.""" + + def __init__(self): + self.données = {} + self.historique = [] + + def définir_préférence(self, clé: str, valeur) -> bool: + """ + Définit une préférence utilisateur. + + Args: + clé: Identifiant de la préférence + valeur: Valeur à enregistrer + + Returns: + True si la préférence a été définie avec succès + """ + try: + self.données[clé] = valeur + self.historique.append({ + "action": "définition", + "clé": clé, + "horodatage": "2024-01-01T12:00:00Z" + }) + return True + except Exception as e: + print(f"Error setting preference: {e}") + return False + + def obtenir_préférence(self, clé: str) -> Optional: + """Récupère une préférence par sa clé.""" + return self.données.get(clé) + + def exporter_données(self) -> str: + """Exporte les données en JSON UTF-8.""" + return json.dumps(self.données, ensure_ascii=False, indent=2) + +# Configuration par défaut avec caractères UTF-8 +CONFIG_DÉFAUT = { + "langue": "français", + "région": "France", + "thème": "sombre", + "notifications": "activées" +} + +def créer_gestionnaire() -> GestionnairePreferences: + """Crée une instance du gestionnaire.""" + gestionnaire = GestionnairePreferences() + + # Application de la configuration par défaut + for clé, valeur in CONFIG_DÉFAUT.items(): + gestionnaire.définir_préférence(clé, valeur) + + return gestionnaire + +if __name__ == "__main__": + # Test d'utilisation + gestionnaire = créer_gestionnaire() + print("Gestionnaire créé avec succès! 🎉") + print(f"Données: {gestionnaire.exporter_données()}") +''' + + # Test writing and reading UTF-8 + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".py", delete=False) as f: + f.write(french_content) + temp_file = f.name + + try: + # Test reading + with open(temp_file, "r", encoding="utf-8") as f: + read_content = f.read() + + # Checks + assert read_content == french_content, "Altered UTF-8 content" + + # Check specific terms + utf8_terms = [ + "préférences", + "Développé", + "Équipe", + "création", + "données", + "définir_préférence", + "horodatage", + "Récupère", + "français", + "activées", + "créer_gestionnaire", + "succès", + ] + + for term in utf8_terms: + assert term in read_content, f"Missing UTF-8 term: {term}" + + print(" ✅ UTF-8 file handling: SUCCESS") + + finally: + # Cleanup + os.unlink(temp_file) + + +def test_mcp_tools_integration(): + """Test MCP tools integration with UTF-8.""" + print(" Testing MCP tools integration...") + + # Simulation of MCP tool response + def simulate_mcp_tool_response(): + """Simulate MCP tool response with UTF-8 content.""" + response_data = { + "status": "success", + "content_type": "markdown", + "content": """# Analysis Completed Successfully ✅ + +## Analysis Summary + +The architectural analysis of the project has been **successfully** completed. Here are the main results: + +### 🎯 Achieved Goals +- ✅ Complete code review +- ✅ Identification of performance issues +- ✅ Improvement recommendations generated + +### 📊 Analyzed Metrics +| Metric | Value | Status | +|--------|-------|--------| +| Cyclomatic complexity | 12 | 🟡 Acceptable | +| Test coverage | 85% | 🟢 Good | +| External dependencies | 23 | 🟠 To be reviewed | + +### 🔍 Identified Issues + +#### 🔴 Critical +No critical issues detected. + +#### 🟠 High +1. **Query performance**: Optimization needed +2. **Memory management**: Potential leaks detected + +#### 🟡 Medium +1. **Documentation**: Some functions lack comments +2. **Unit tests**: Coverage to be improved + +### 🚀 Priority Recommendations + +1. **DB Optimization**: Implement Redis cache +2. **Refactoring**: Separate responsibilities +3. **Documentation**: Add missing docstrings +4. **Tests**: Increase coverage to 90%+ + +### 📈 Next Steps + +- [ ] Implement caching system +- [ ] Refactor identified modules +- [ ] Complete documentation +- [ ] Run regression tests + +--- +*Analysis automatically generated by MCP Zen* 🤖 +""", + "metadata": { + "tool_name": "analyze", + "execution_time": 2.5, + "locale": "fr-FR", + "timestamp": "2024-01-01T12:00:00Z", + "analysis_summary": { + "files_analyzed": 15, + "issues_found": 4, + "recommendations": 4, + "overall_score": "B+ (Good level)", + }, + }, + "continuation_offer": { + "continuation_id": "analysis-123", + "note": "In-depth analysis available with more details", + }, + } + + # Serialization with ensure_ascii=False + json_response = json.dumps(response_data, ensure_ascii=False, indent=2) + + # UTF-8 checks + utf8_checks = [ + "Terminée", + "Succès", + "Résumé", + "terminée", + "Atteints", + "Révision", + "problèmes", + "générées", + "Métriques", + "Identifiés", + "détecté", + "Élevé", + "nécessaire", + "détectées", + "améliorer", + "Prioritaires", + "responsabilités", + "Étapes", + "régression", + "générée", + "détails", + ] + + for term in utf8_checks: + assert term in json_response, f"Missing UTF-8 term: {term}" + + # Emoji check + emojis = ["✅", "🎯", "📊", "🟡", "🟢", "🟠", "🔍", "🔴", "🚀", "📈", "🤖"] + for emoji in emojis: + assert emoji in json_response, f"Missing emoji: {emoji}" + + # Test parsing + parsed = json.loads(json_response) + assert parsed["status"] == "success" + assert "Terminée" in parsed["content"] + assert parsed["metadata"]["locale"] == "fr-FR" + + return json_response + + # Test simulation + response = simulate_mcp_tool_response() + assert len(response) > 1000, "MCP response too short" + + print(" ✅ MCP tools integration: SUCCESS") + + +def run_unit_tests(): + """Run unit tests.""" + print(" Running unit tests...") + + # List of test files to run + test_files = ["test_utf8_localization.py", "test_provider_utf8.py", "test_workflow_utf8.py"] + + current_dir = Path(__file__).parent + test_results = [] + + for test_file in test_files: + test_path = current_dir / test_file + if test_path.exists(): + print(f" 📝 Running {test_file}...") + try: + # Test execution + result = subprocess.run( + [sys.executable, "-m", "unittest", test_file.replace(".py", ""), "-v"], + cwd=current_dir, + capture_output=True, + text=True, + timeout=60, + ) + + if result.returncode == 0: + print(f" ✅ {test_file}: SUCCESS") + test_results.append((test_file, "SUCCESS")) + else: + print(f" ❌ {test_file}: FAILURE") + print(f" Error: {result.stderr[:200]}...") + test_results.append((test_file, "FAILURE")) + + except subprocess.TimeoutExpired: + print(f" ⏰ {test_file}: TIMEOUT") + test_results.append((test_file, "TIMEOUT")) + except Exception as e: + print(f" 💥 {test_file}: ERROR - {e}") + test_results.append((test_file, "ERROR")) + else: + print(f" ⚠️ {test_file}: NOT FOUND") + test_results.append((test_file, "NOT FOUND")) + + # Test summary + print("\n 📋 Unit test summary:") + for test_file, status in test_results: + status_emoji = {"SUCCESS": "✅", "FAILURE": "❌", "TIMEOUT": "⏰", "ERROR": "💥", "NOT FOUND": "⚠️"}.get( + status, "❓" + ) + print(f" {status_emoji} {test_file}: {status}") + + +def main(): + """Main function.""" + print("🇫🇷 UTF-8 Integration Test - Zen MCP Server") + print("=" * 60) + + try: + run_utf8_integration_tests() + print("\n🎉 SUCCESS: All UTF-8 integration tests passed!") + print("🚀 Zen MCP server fully supports French localization!") + return 0 + + except AssertionError as e: + print(f"\n❌ FAILURE: Assertion test failed: {e}") + return 1 + + except Exception as e: + print(f"\n💥 ERROR: Unexpected exception: {e}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_provider_utf8.py b/tests/test_provider_utf8.py new file mode 100644 index 0000000..ff95f12 --- /dev/null +++ b/tests/test_provider_utf8.py @@ -0,0 +1,352 @@ +""" +Unit tests to validate UTF-8 encoding in providers +and integration with language models. +""" + +import json +import os +import unittest +from unittest.mock import Mock, patch + +import pytest + +from providers.base import ModelProvider, ProviderType +from providers.gemini import GeminiModelProvider +from providers.openai_compatible import OpenAICompatibleProvider +from providers.openai_provider import OpenAIModelProvider + + +class TestProviderUTF8Encoding(unittest.TestCase): + """Tests for UTF-8 encoding in providers.""" + + def setUp(self): + """Test setup.""" + self.original_locale = os.getenv("LOCALE") + + def tearDown(self): + """Cleanup after tests.""" + if self.original_locale is not None: + os.environ["LOCALE"] = self.original_locale + else: + os.environ.pop("LOCALE", None) + + def test_base_provider_utf8_support(self): + """Test that the base provider supports UTF-8.""" + provider = ModelProvider(api_key="test") + + # Test with UTF-8 characters + test_text = "Développement en français avec émojis 🚀" + tokens = provider.count_tokens(test_text, "test-model") + + # Should return a valid number (character-based estimate) + self.assertIsInstance(tokens, int) + self.assertGreater(tokens, 0) + + @patch("google.generativeai.GenerativeModel") + def test_gemini_provider_utf8_request(self, mock_model_class): + """Test that the Gemini provider handles UTF-8 correctly.""" + # Mock Gemini response + mock_response = Mock() + mock_response.text = "Response in French with accents: créé, développé, préféré 🎉" + mock_response.usage_metadata = Mock() + mock_response.usage_metadata.prompt_token_count = 10 + mock_response.usage_metadata.candidates_token_count = 15 + mock_response.usage_metadata.total_token_count = 25 + + mock_model = Mock() + mock_model.generate_content.return_value = mock_response + mock_model_class.return_value = mock_model + + # Test Gemini provider + provider = GeminiModelProvider(api_key="test-key") + + # Request with UTF-8 characters + response = provider.generate_content( + prompt="Can you explain software development?", + model_name="gemini-2.5-flash", + system_prompt="Reply in French with emojis.", + ) + + # Checks + self.assertIsNotNone(response) + self.assertIn("French", response.content) + self.assertIn("🎉", response.content) + + # Check that the request contains UTF-8 characters + mock_model.generate_content.assert_called_once() + call_args = mock_model.generate_content.call_args + parts = call_args[0][0] # First argument (parts) + + # Check for UTF-8 content in the request + request_content = str(parts) + self.assertIn("développement", request_content) + + @patch("openai.OpenAI") + def test_openai_provider_utf8_logging(self, mock_openai_class): + """Test that the OpenAI provider logs UTF-8 correctly.""" + # Mock OpenAI response + mock_response = Mock() + mock_response.choices = [Mock()] + mock_response.choices[0].message = Mock() + mock_response.choices[0].message.content = "Python code created successfully! ✅" + mock_response.usage = Mock() + mock_response.usage.prompt_tokens = 20 + mock_response.usage.completion_tokens = 10 + mock_response.usage.total_tokens = 30 + + mock_client = Mock() + mock_client.chat.completions.create.return_value = mock_response + mock_openai_class.return_value = mock_client + + # Test OpenAI provider + provider = OpenAIModelProvider(api_key="test-key") + + # Test with UTF-8 logging + with patch("logging.info") as mock_logging: + response = provider.generate_content( + prompt="Generate Python code to process data", + model_name="gpt-4", + system_prompt="You are an expert Python developer.", + ) + + # Response checks + self.assertIsNotNone(response) + self.assertIn("created", response.content) + self.assertIn("✅", response.content) + + @patch("openai.OpenAI") + def test_openai_compatible_o3_pro_utf8(self, mock_openai_class): + """Specific test for o3-pro with /responses endpoint and UTF-8.""" + # Mock o3-pro response + mock_response = Mock() + mock_response.output = Mock() + mock_response.output.content = [Mock()] + mock_response.output.content[0].type = "output_text" + mock_response.output.content[0].text = "Analysis complete: code is well structured! 🎯" + mock_response.usage = Mock() + mock_response.usage.input_tokens = 50 + mock_response.usage.output_tokens = 25 + mock_response.model = "o3-pro-2025-06-10" + mock_response.id = "test-id" + mock_response.created_at = 1234567890 + + mock_client = Mock() + mock_client.responses.create.return_value = mock_response + mock_openai_class.return_value = mock_client + + # Test OpenAI Compatible provider with o3-pro + provider = OpenAICompatibleProvider(api_key="test-key", base_url="https://api.openai.com/v1") + + # Test with UTF-8 logging for o3-pro + with patch("logging.info") as mock_logging: + response = provider.generate_content( + prompt="Analyze this Python code for issues", + model_name="o3-pro-2025-06-10", + system_prompt="You are a code review expert.", + ) + + # Response checks + self.assertIsNotNone(response) + self.assertIn("complete", response.content) + self.assertIn("🎯", response.content) + + # Check that logging was called with ensure_ascii=False + mock_logging.assert_called() + log_calls = [call for call in mock_logging.call_args_list if "API request payload" in str(call)] + self.assertTrue(len(log_calls) > 0, "No API payload log found") + + def test_provider_type_enum_utf8_safe(self): + """Test that ProviderType enum is UTF-8 safe.""" + # Test all provider types + provider_types = list(ProviderType) + + for provider_type in provider_types: + # Test JSON serialization + data = {"provider": provider_type.value, "message": "UTF-8 test: emojis 🚀"} + json_str = json.dumps(data, ensure_ascii=False) + + # Checks + self.assertIn(provider_type.value, json_str) + self.assertIn("emojis", json_str) + self.assertIn("🚀", json_str) + + # Test deserialization + parsed = json.loads(json_str) + self.assertEqual(parsed["provider"], provider_type.value) + self.assertEqual(parsed["message"], "UTF-8 test: emojis 🚀") + + def test_model_response_utf8_serialization(self): + """Test UTF-8 serialization of model responses.""" + from providers.base import ModelResponse + + # Create a response with UTF-8 characters + response = ModelResponse( + content="Development successful! Code generated successfully. 🎉✅", + usage={"input_tokens": 10, "output_tokens": 15, "total_tokens": 25}, + model_name="test-model", + friendly_name="Test Model", + provider=ProviderType.OPENAI, + metadata={"created": "2024-01-01", "developer": "Test", "emojis": "🚀🎯🔥"}, + ) + + # Test serialization + response_dict = response.to_dict() + json_str = json.dumps(response_dict, ensure_ascii=False, indent=2) + + # Checks + self.assertIn("Development", json_str) + self.assertIn("successful", json_str) + self.assertIn("generated", json_str) + self.assertIn("🎉", json_str) + self.assertIn("✅", json_str) + self.assertIn("created", json_str) + self.assertIn("developer", json_str) + self.assertIn("🚀", json_str) + + # Test deserialization + parsed = json.loads(json_str) + self.assertEqual(parsed["content"], response.content) + self.assertEqual(parsed["friendly_name"], "Test Model") + + def test_error_handling_with_utf8(self): + """Test error handling with UTF-8 characters.""" + provider = ModelProvider(api_key="test") + + # Test validation with UTF-8 error message + with self.assertRaises(ValueError) as context: + provider.validate_parameters("", -1.0) # Invalid temperature + + error_message = str(context.exception) + # Error message may contain UTF-8 characters + self.assertIsInstance(error_message, str) + + def test_temperature_handling_utf8_locale(self): + """Test temperature handling with UTF-8 locale.""" + # Set French locale + os.environ["LOCALE"] = "fr-FR" + + provider = ModelProvider(api_key="test") + + # Test different temperatures + test_temps = [0.0, 0.5, 1.0, 1.5, 2.0] + + for temp in test_temps: + try: + provider.validate_parameters("gpt-4", temp) + # If no exception, temperature is valid + self.assertLessEqual(temp, 2.0) + except ValueError: + # If exception, temperature must be > 2.0 + self.assertGreater(temp, 2.0) + + def test_provider_registry_utf8(self): + """Test that the provider registry handles UTF-8.""" + from providers.registry import ModelProviderRegistry + + # Test listing providers with UTF-8 descriptions + providers = ModelProviderRegistry.get_available_providers() + + # Should contain valid providers + self.assertGreater(len(providers), 0) + + # Test serialization + provider_data = { + "providers": [p.value for p in providers], + "description": "Available providers for development 🚀", + } + + json_str = json.dumps(provider_data, ensure_ascii=False) + + # Checks + self.assertIn("development", json_str) + self.assertIn("🚀", json_str) + + # Test parsing + parsed = json.loads(json_str) + self.assertEqual(parsed["description"], provider_data["description"]) + + +class TestLocaleModelIntegration(unittest.TestCase): + """Integration tests between locale and models.""" + + def setUp(self): + """Integration test setup.""" + self.original_locale = os.getenv("LOCALE") + + def tearDown(self): + """Cleanup after integration tests.""" + if self.original_locale is not None: + os.environ["LOCALE"] = self.original_locale + else: + os.environ.pop("LOCALE", None) + + def test_system_prompt_enhancement_french(self): + """Test system prompt enhancement with French locale.""" + # Set to French + os.environ["LOCALE"] = "fr-FR" + + provider = ModelProvider(api_key="test") + base_prompt = "You are a helpful coding assistant." + + # Test prompt enhancement + enhanced_prompt = provider.enhance_system_prompt(base_prompt) + + # Checks + self.assertIn("fr-FR", enhanced_prompt) + self.assertIn(base_prompt, enhanced_prompt) + + def test_system_prompt_enhancement_multiple_locales(self): + """Test enhancement with different locales.""" + provider = ModelProvider(api_key="test") + base_prompt = "You are a helpful assistant." + + locales = ["fr-FR", "es-ES", "de-DE", "it-IT", "pt-BR", "ja-JP", "zh-CN"] + + for locale in locales: + os.environ["LOCALE"] = locale + enhanced_prompt = provider.enhance_system_prompt(base_prompt) + + # Locale-specific checks + self.assertIn(locale, enhanced_prompt) + self.assertIn(base_prompt, enhanced_prompt) + + # Test JSON serialization + prompt_data = {"system_prompt": enhanced_prompt, "locale": locale} + json_str = json.dumps(prompt_data, ensure_ascii=False) + + # Should parse without error + parsed = json.loads(json_str) + self.assertEqual(parsed["locale"], locale) + + def test_model_name_resolution_utf8(self): + """Test model name resolution with UTF-8.""" + provider = ModelProvider(api_key="test") + + # Test with different model names + model_names = ["gpt-4", "gemini-2.5-flash", "claude-3-opus", "o3-pro-2025-06-10"] + + for model_name in model_names: + # Test resolution + resolved = provider._resolve_model_name(model_name) + self.assertIsInstance(resolved, str) + + # Test serialization with UTF-8 metadata + model_data = { + "model": resolved, + "description": f"Model {model_name} - advanced development 🚀", + "capabilities": ["generation", "review", "creation"], + } + + json_str = json.dumps(model_data, ensure_ascii=False) + + # Checks + self.assertIn("development", json_str) + self.assertIn("generation", json_str) + self.assertIn("review", json_str) + self.assertIn("creation", json_str) + self.assertIn("🚀", json_str) + + +if __name__ == "__main__": + # Test configuration + pytest.main([__file__, "-v", "--tb=short"]) diff --git a/tests/test_refactor.py b/tests/test_refactor.py index 485994b..9b8cf93 100644 --- a/tests/test_refactor.py +++ b/tests/test_refactor.py @@ -46,7 +46,8 @@ class TestRefactorTool: ], "priority_sequence": ["refactor-001"], "next_actions_for_claude": [], - } + }, + ensure_ascii=False ) from unittest.mock import Mock diff --git a/tests/test_utf8_localization.py b/tests/test_utf8_localization.py new file mode 100644 index 0000000..14bb786 --- /dev/null +++ b/tests/test_utf8_localization.py @@ -0,0 +1,427 @@ +""" +Unit tests to validate UTF-8 localization and encoding +of French characters. + +These tests check: +1. Language instruction generation according to LOCALE +2. UTF-8 encoding with json.dumps(ensure_ascii=False) +3. French characters and emojis are displayed correctly +4. MCP tools return localized content +""" + +import json +import os +import tempfile +import unittest +from unittest.mock import Mock, patch + +import pytest + +from tools.chat import ChatTool +from tools.codereview import CodereviewTool +from tools.shared.base_tool import BaseTool + + +class TestUTF8Localization(unittest.TestCase): + """Tests for UTF-8 localization and French character encoding.""" + + def setUp(self): + """Test setup.""" + self.original_locale = os.getenv("LOCALE") + + def tearDown(self): + """Cleanup after tests.""" + if self.original_locale is not None: + os.environ["LOCALE"] = self.original_locale + else: + os.environ.pop("LOCALE", None) + + def test_language_instruction_generation_french(self): + """Test language instruction generation for French.""" + # Set LOCALE to French + os.environ["LOCALE"] = "fr-FR" + + # Test get_language_instruction method + tool = BaseTool(api_key="test") + instruction = tool.get_language_instruction() + + # Checks + self.assertIsInstance(instruction, str) + self.assertIn("fr-FR", instruction) + self.assertTrue(instruction.endswith("\n\n")) + + def test_language_instruction_generation_english(self): + """Test language instruction generation for English.""" + # Set LOCALE to English + os.environ["LOCALE"] = "en-US" + + tool = BaseTool(api_key="test") + instruction = tool.get_language_instruction() + + # Checks + self.assertIsInstance(instruction, str) + self.assertIn("en-US", instruction) + self.assertTrue(instruction.endswith("\n\n")) + + def test_language_instruction_empty_locale(self): + """Test with empty LOCALE.""" + # Set LOCALE to empty + os.environ["LOCALE"] = "" + + tool = BaseTool(api_key="test") + instruction = tool.get_language_instruction() + + # Should return empty string + self.assertEqual(instruction, "") + + def test_language_instruction_no_locale(self): + """Test with no LOCALE variable set.""" + # Remove LOCALE + os.environ.pop("LOCALE", None) + + tool = BaseTool(api_key="test") + instruction = tool.get_language_instruction() + + # Should return empty string + self.assertEqual(instruction, "") + + def test_json_dumps_utf8_encoding(self): + """Test that json.dumps uses ensure_ascii=False for UTF-8.""" + # Test data with French characters and emojis + test_data = { + "status": "succès", + "message": "Tâche terminée avec succès", + "details": { + "créé": "2024-01-01", + "développeur": "Jean Dupont", + "préférences": ["français", "développement"], + "emojis": "🔴 🟠 🟡 🟢 ✅ ❌", + }, + } + + # Test with ensure_ascii=False (correct) + json_correct = json.dumps(test_data, ensure_ascii=False, indent=2) + + # Check that UTF-8 characters are preserved + self.assertIn("succès", json_correct) + self.assertIn("terminée", json_correct) + self.assertIn("créé", json_correct) + self.assertIn("développeur", json_correct) + self.assertIn("préférences", json_correct) + self.assertIn("français", json_correct) + self.assertIn("développement", json_correct) + self.assertIn("🔴", json_correct) + self.assertIn("🟢", json_correct) + self.assertIn("✅", json_correct) + + # Check that characters are NOT escaped + self.assertNotIn("\\u", json_correct) + self.assertNotIn("\\ud83d", json_correct) + + def test_json_dumps_ascii_encoding_comparison(self): + """Test comparison between ensure_ascii=True and False.""" + test_data = {"message": "Développement réussi! 🎉"} + + # With ensure_ascii=True (old, incorrect behavior) + json_escaped = json.dumps(test_data, ensure_ascii=True) + + # With ensure_ascii=False (new, correct behavior) + json_utf8 = json.dumps(test_data, ensure_ascii=False) + + # Checks + self.assertIn("\\u", json_escaped) # Characters are escaped + self.assertNotIn("é", json_escaped) # UTF-8 characters are escaped + + self.assertNotIn("\\u", json_utf8) # No escaped characters + self.assertIn("é", json_utf8) # UTF-8 characters preserved + self.assertIn("🎉", json_utf8) # Emojis preserved + + @patch("tools.shared.base_tool.BaseTool.get_model_provider") + def test_chat_tool_french_response(self, mock_get_provider): + """Test that the chat tool returns a response in French.""" + # Set to French + os.environ["LOCALE"] = "fr-FR" + + # Mock provider + mock_provider = Mock() + mock_provider.get_provider_type.return_value = Mock(value="test") + mock_provider.generate_content.return_value = Mock( + content="Bonjour! Je peux vous aider avec vos tâches de développement.", + usage={}, + model_name="test-model", + metadata={}, + ) + mock_get_provider.return_value = mock_provider + + # Test chat tool + chat_tool = ChatTool() + result = chat_tool.execute({"prompt": "Peux-tu m'aider?", "model": "test-model"}) + + # Checks + self.assertIsNotNone(result) + self.assertEqual(len(result), 1) + + # Parse JSON response + response_data = json.loads(result[0].text) + + # Check that response contains French content + self.assertIn("status", response_data) + self.assertIn("content", response_data) + + # Check that language instruction was added + mock_provider.generate_content.assert_called_once() + call_args = mock_provider.generate_content.call_args + system_prompt = call_args.kwargs.get("system_prompt", "") + self.assertIn("fr-FR", system_prompt) + + def test_french_characters_in_file_content(self): + """Test reading and writing files with French characters.""" + # Test content with French characters + test_content = """ +# System configuration +# Created by: Lead Developer +# Creation date: December 15, 2024 + +def process_data(preferences, parameters): + ''' + Processes data according to user preferences. + + Args: + preferences: User preferences dictionary + parameters: Configuration parameters + + Returns: + Processing result + ''' + return "Processing completed successfully! ✅" + +# Helper functions +def generate_report(): + '''Generates a summary report.''' + return { + "status": "success", + "data": "Report generated", + "emojis": "📊 📈 📉" + } +""" + + # Test writing and reading + with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8", delete=False) as f: + f.write(test_content) + temp_file = f.name + + try: + # Read file + with open(temp_file, "r", encoding="utf-8") as f: + read_content = f.read() + + # Checks + self.assertEqual(read_content, test_content) + self.assertIn("Lead Developer", read_content) + self.assertIn("Creation", read_content) + self.assertIn("data", read_content) + self.assertIn("preferences", read_content) + self.assertIn("parameters", read_content) + self.assertIn("completed", read_content) + self.assertIn("successfully", read_content) + self.assertIn("✅", read_content) + self.assertIn("success", read_content) + self.assertIn("generated", read_content) + self.assertIn("📊", read_content) + + finally: + # Cleanup + os.unlink(temp_file) + + def test_system_prompt_integration_french(self): + """Test integration of language instruction in system prompts.""" + # Set to French + os.environ["LOCALE"] = "fr-FR" + + tool = BaseTool(api_key="test") + base_prompt = "You are a helpful assistant." + + # Test adding language instruction + enhanced_prompt = tool.add_language_instruction(base_prompt) + + # Checks + self.assertIn("fr-FR", enhanced_prompt) + self.assertIn(base_prompt, enhanced_prompt) + self.assertTrue(enhanced_prompt.startswith("Always respond in fr-FR")) + + def test_system_prompt_integration_no_locale(self): + """Test integration with no LOCALE set.""" + # No LOCALE + os.environ.pop("LOCALE", None) + + tool = BaseTool(api_key="test") + base_prompt = "You are a helpful assistant." + + # Test adding language instruction + enhanced_prompt = tool.add_language_instruction(base_prompt) + + # Should return original prompt unchanged + self.assertEqual(enhanced_prompt, base_prompt) + + def test_unicode_normalization(self): + """Test Unicode normalization for accented characters.""" + # Test with different Unicode encodings + test_cases = [ + "café", # e + acute accent combined + "café", # e with precomposed acute accent + "naïf", # i + diaeresis + "coeur", # oe ligature + "été", # e + acute accent + ] + + for text in test_cases: + # Test that json.dumps preserves characters + json_output = json.dumps({"text": text}, ensure_ascii=False) + self.assertIn(text, json_output) + + # Parse and check + parsed = json.loads(json_output) + self.assertEqual(parsed["text"], text) + + def test_emoji_preservation(self): + """Test emoji preservation in JSON encoding.""" + # Emojis used in Zen MCP tools + emojis = [ + "🔴", # Critical + "🟠", # High + "🟡", # Medium + "🟢", # Low + "✅", # Success + "❌", # Error + "⚠️", # Warning + "📊", # Charts + "🎉", # Celebration + "🚀", # Rocket + "🇫🇷", # French flag + ] + + test_data = {"emojis": emojis, "message": " ".join(emojis)} + + # Test with ensure_ascii=False + json_output = json.dumps(test_data, ensure_ascii=False) + + # Checks + for emoji in emojis: + self.assertIn(emoji, json_output) + + # No escaped characters + self.assertNotIn("\\u", json_output) + + # Test parsing + parsed = json.loads(json_output) + self.assertEqual(parsed["emojis"], emojis) + self.assertEqual(parsed["message"], " ".join(emojis)) + + +class TestLocalizationIntegration(unittest.TestCase): + """Integration tests for localization with real tools.""" + + def setUp(self): + """Integration test setup.""" + self.original_locale = os.getenv("LOCALE") + + def tearDown(self): + """Cleanup after integration tests.""" + if self.original_locale is not None: + os.environ["LOCALE"] = self.original_locale + else: + os.environ.pop("LOCALE", None) + + @patch("tools.shared.base_tool.BaseTool.get_model_provider") + def test_codereview_tool_french_locale(self, mock_get_provider): + """Test that the codereview tool uses French localization.""" + # Set to French + os.environ["LOCALE"] = "fr-FR" + + # Mock provider with French response + mock_provider = Mock() + mock_provider.get_provider_type.return_value = Mock(value="test") + mock_provider.generate_content.return_value = Mock( + content=json.dumps( + {"status": "analysis_complete", "raw_analysis": "Code review completed. No critical issues found. 🟢"}, + ensure_ascii=False, + ), + usage={}, + model_name="test-model", + metadata={}, + ) + mock_get_provider.return_value = mock_provider + + # Test codereview tool + codereview_tool = CodereviewTool() + result = codereview_tool.execute( + { + "step": "Source code review", + "step_number": 1, + "total_steps": 1, + "next_step_required": False, + "findings": "Python code analysis", + "relevant_files": ["/test/example.py"], + "model": "test-model", + } + ) + + # Checks + self.assertIsNotNone(result) + self.assertEqual(len(result), 1) + + # Parse JSON response - should be valid UTF-8 + response_text = result[0].text + response_data = json.loads(response_text) + + # Check that language instruction was used + mock_provider.generate_content.assert_called() + call_args = mock_provider.generate_content.call_args + system_prompt = call_args.kwargs.get("system_prompt", "") + self.assertIn("fr-FR", system_prompt) + + # Check that response contains UTF-8 characters + if "expert_analysis" in response_data: + expert_analysis = response_data["expert_analysis"] + if "raw_analysis" in expert_analysis: + analysis = expert_analysis["raw_analysis"] + # Should contain French characters + self.assertTrue( + any(char in analysis for char in ["é", "è", "à", "ç", "ê", "û", "î", "ô"]) or "🟢" in analysis + ) + + def test_multiple_locales_switching(self): + """Test switching locales during execution.""" + tool = BaseTool(api_key="test") + + # French + os.environ["LOCALE"] = "fr-FR" + instruction_fr = tool.get_language_instruction() + self.assertIn("fr-FR", instruction_fr) + + # English + os.environ["LOCALE"] = "en-US" + instruction_en = tool.get_language_instruction() + self.assertIn("en-US", instruction_en) + + # Spanish + os.environ["LOCALE"] = "es-ES" + instruction_es = tool.get_language_instruction() + self.assertIn("es-ES", instruction_es) + + # Chinese + os.environ["LOCALE"] = "zh-CN" + instruction_zh = tool.get_language_instruction() + self.assertIn("zh-CN", instruction_zh) + + # Check that all instructions are different + instructions = [instruction_fr, instruction_en, instruction_es, instruction_zh] + for i, inst1 in enumerate(instructions): + for j, inst2 in enumerate(instructions): + if i != j: + self.assertNotEqual(inst1, inst2) + + +if __name__ == "__main__": + # Test configuration + pytest.main([__file__, "-v", "--tb=short"]) diff --git a/tests/test_workflow_utf8.py b/tests/test_workflow_utf8.py new file mode 100644 index 0000000..83da2d7 --- /dev/null +++ b/tests/test_workflow_utf8.py @@ -0,0 +1,456 @@ +""" +Unit tests to validate UTF-8 encoding in workflow tools +and the generation of properly encoded JSON responses. +""" + +import json +import os +import tempfile +import unittest +from unittest.mock import Mock, patch + +from tools.analyze import AnalyzeTool +from tools.codereview import CodereviewTool +from tools.debug import DebugIssueTool + + +class TestWorkflowToolsUTF8(unittest.TestCase): + """Tests for UTF-8 encoding in workflow tools.""" + + def setUp(self): + """Test setup.""" + self.original_locale = os.getenv("LOCALE") + # Default to French for tests + os.environ["LOCALE"] = "fr-FR" + + def tearDown(self): + """Cleanup after tests.""" + if self.original_locale is not None: + os.environ["LOCALE"] = self.original_locale + else: + os.environ.pop("LOCALE", None) + + def test_workflow_json_response_structure(self): + """Test the structure of JSON responses from workflow tools.""" + # Test with analysis tool + analyze_tool = AnalyzeTool() + + # Mock response with UTF-8 characters + test_response = { + "status": "pause_for_analysis", + "step_number": 1, + "total_steps": 3, + "next_step_required": True, + "findings": "Code analysis reveals performance issues 🔍", + "files_checked": ["/src/main.py"], + "relevant_files": ["/src/main.py"], + "issues_found": [ + {"severity": "high", "description": "Function too complex - refactoring needed"} + ], + "investigation_required": True, + "required_actions": ["Review code dependencies", "Analyze architectural patterns"], + } + + # Test JSON serialization with ensure_ascii=False + json_str = json.dumps(test_response, indent=2, ensure_ascii=False) + + # UTF-8 checks + self.assertIn("révèle", json_str) + self.assertIn("problèmes", json_str) + self.assertIn("nécessaire", json_str) + self.assertIn("dépendances", json_str) + self.assertIn("🔍", json_str) + + # No escaped characters + self.assertNotIn("\\u", json_str) + + # Test parsing + parsed = json.loads(json_str) + self.assertEqual(parsed["findings"], test_response["findings"]) + self.assertEqual(len(parsed["issues_found"]), 1) + self.assertIn("nécessaire", parsed["issues_found"][0]["description"]) + + @patch("tools.shared.base_tool.BaseTool.get_model_provider") + def test_analyze_tool_utf8_response(self, mock_get_provider): + """Test that the analyze tool returns correct UTF-8 responses.""" + # Mock provider + mock_provider = Mock() + mock_provider.get_provider_type.return_value = Mock(value="test") + mock_provider.generate_content.return_value = Mock( + content="Architectural analysis complete. Recommendations: improve modularity.", + usage={}, + model_name="test-model", + metadata={}, + ) + mock_get_provider.return_value = mock_provider + + # Test the tool + analyze_tool = AnalyzeTool() + result = analyze_tool.execute( + { + "step": "Analyze system architecture to identify issues", + "step_number": 1, + "total_steps": 2, + "next_step_required": True, + "findings": "Starting architectural analysis of Python code", + "relevant_files": ["/test/main.py"], + "model": "test-model", + } + ) + + # Checks + self.assertIsNotNone(result) + self.assertEqual(len(result), 1) + + # Parse the response - must be valid UTF-8 JSON + response_text = result[0].text + response_data = json.loads(response_text) + + # Structure checks + self.assertIn("status", response_data) + self.assertIn("step_number", response_data) + + # Check that the French instruction was added + mock_provider.generate_content.assert_called() + call_args = mock_provider.generate_content.call_args + system_prompt = call_args.kwargs.get("system_prompt", "") + self.assertIn("fr-FR", system_prompt) + + @patch("tools.shared.base_tool.BaseTool.get_model_provider") + def test_codereview_tool_french_findings(self, mock_get_provider): + """Test that the codereview tool produces findings in French.""" + # Mock with analysis in French + mock_provider = Mock() + mock_provider.get_provider_type.return_value = Mock(value="test") + mock_provider.supports_thinking_mode.return_value = False + mock_provider.generate_content.return_value = Mock( + content=json.dumps( + { + "status": "analysis_complete", + "raw_analysis": """ +🔴 CRITIQUE: Aucun problème critique trouvé. + +🟠 ÉLEVÉ: Fichier example.py:42 - Fonction trop complexe +→ Problème: La fonction process_data() contient trop de responsabilités +→ Solution: Décomposer en fonctions plus petites et spécialisées + +🟡 MOYEN: Gestion d'erreurs insuffisante +→ Problème: Plusieurs fonctions n'ont pas de gestion d'erreurs appropriée +→ Solution: Ajouter des try-catch et validation des paramètres + +✅ Points positifs: +• Code bien commenté et lisible +• Nomenclature cohérente +• Tests unitaires présents +""", + }, + ensure_ascii=False, + ), + usage={}, + model_name="test-model", + metadata={}, + ) + mock_get_provider.return_value = mock_provider + + # Test the tool + codereview_tool = CodereviewTool() + result = codereview_tool.execute( + { + "step": "Complete review of Python code", + "step_number": 1, + "total_steps": 1, + "next_step_required": False, + "findings": "Code review complete", + "relevant_files": ["/test/example.py"], + "model": "test-model", + } + ) + + # Checks + self.assertIsNotNone(result) + response_text = result[0].text + response_data = json.loads(response_text) + + # Check UTF-8 characters in analysis + if "expert_analysis" in response_data: + analysis = response_data["expert_analysis"]["raw_analysis"] + # Vérification de caractères français + # Check for French characters + self.assertIn("ÉLEVÉ", analysis)is) + self.assertIn("problème", analysis)sis) + self.assertIn("spécialisées", analysis) + self.assertIn("appropriée", analysis) + self.assertIn("paramètres", analysis) + self.assertIn("présents", analysis) + # Vérification d'emojis + # Check for emojislysis) + self.assertIn("🔴", analysis) + self.assertIn("🟠", analysis) + self.assertIn("🟡", analysis) + self.assertIn("✅", analysis) + @patch("tools.shared.base_tool.BaseTool.get_model_provider") + @patch("tools.shared.base_tool.BaseTool.get_model_provider")vider): + def test_debug_tool_french_error_analysis(self, mock_get_provider): + """Test that the debug tool analyzes errors in French.""" + # Mock providerck() + mock_provider = Mock()ider_type.return_value = Mock(value="test") + mock_provider.get_provider_type.return_value = Mock(value="test") + mock_provider.generate_content.return_value = Mock(n définie. Cause probable: import manquant.", + content="Error analyzed: variable 'données' not defined. Probable cause: missing import.", + usage={},e="test-model", + model_name="test-model", + metadata={}, + )ock_get_provider.return_value = mock_provider + mock_get_provider.return_value = mock_provider + # Test de l'outil debug + # Test the debug toolTool() + debug_tool = DebugIssueTool() + result = debug_tool.execute( + { "step": "Analyser l'erreur NameError dans le fichier de traitement des données", + "step": "Analyze NameError in data processing file", + "step_number": 1, + "total_steps": 2,ed": True, + "next_step_required": True,e lors de l'exécution du script", + "findings": "Error detected during script execution", + "files_checked": ["/src/data_processor.py"],, + "relevant_files": ["/src/data_processor.py"], - import manquant", + "hypothesis": "Variable 'données' not defined - missing import", + "confidence": "medium", + "model": "test-model", + } + ) + # Vérifications + # CheckstNone(result) + self.assertIsNotNone(result)xt + response_text = result[0].textponse_text) + response_data = json.loads(response_text) + # Vérification de la structure de réponse + # Check response structure + self.assertIn("status", response_data)response_data) + self.assertIn("investigation_status", response_data) + # Vérification que les caractères UTF-8 sont préservés + # Check that UTF-8 characters are preservedFalse) + response_str = json.dumps(response_data, ensure_ascii=False) + self.assertIn("données", response_str)) + self.assertIn("détectée", response_str)) + self.assertIn("exécution", response_str) + self.assertIn("définie", response_str) + def test_workflow_mixin_utf8_serialization(self): + def test_workflow_mixin_utf8_serialization(self):lowMixin.""" + """Test UTF-8 serialization in BaseWorkflowMixin.""" + # Simulation of a workflow response with UTF-8 characters + workflow_response = {g_expert_analysis", + "status": "calling_expert_analysis", + "step_number": 2, + "total_steps": 3,ed": True, + "next_step_required": True,", + "continuation_id": "test-id", + "file_context": {y_embedded", + "type": "fully_embedded", + "files_embedded": 2,n": "Contexte optimisé pour l'analyse experte", + "context_optimization": "Context optimized for expert analysis", + },xpert_analysis": { + "expert_analysis": {sis_complete", + "status": "analysis_complete", + "raw_analysis": """ +Complete system analysis reveals: +🎯 **Objectif**: Améliorer les performances +🎯 **Objective**: Improve performancenamique +🔍 **Methodology**: Static and dynamic analysis +📊 **Results**: nérale: satisfaisante + • Overall performance: satisfactoryées + • Possible optimizations: 3 identifiedlog n) + • Algorithmic complexity: O(n²) → O(n log n) +**Recommandations prioritaires**: +**Priority recommendations**:es données +1. Optimize the data sorting functionréquentes +2. Implement a cache for frequent requests +3. Refactor the report generation module +🚀 **Impact attendu**: Amélioration de 40% des performances +🚀 **Expected impact**: 40% improvement in performance +""", }, + },nvestigation_summary": { + "investigation_summary": {rc/performance.py", "/src/cache.py"], + "files_analyzed": ["/src/performance.py", "/src/cache.py"],nt des données", + "key_findings": "Optimizations identified in data processing", + "recommendations": "Implement caching and algorithmic improvement", + }, + } + # Test de sérialisation avec ensure_ascii=False + # Test serialization with ensure_ascii=False=2, ensure_ascii=False) + json_str = json.dumps(workflow_response, indent=2, ensure_ascii=False) + # Vérifications de préservation UTF-8 + # UTF-8 preservation checks + utf8_chars = [ + "révèle",ogie", + "Méthodologie", + "générale",s", + "identifiées",, + "prioritaires", + "données",s", + "fréquentes", + "génération", + "attendu",ion", + "Amélioration", + "identifiées",, + "amélioration", + ] + for char_seq in utf8_chars: + for char_seq in utf8_chars: json_str) + self.assertIn(char_seq, json_str) + # Vérifications d'emojis + # Emoji checks", "🚀"] + emojis = ["🎯", "🔍", "📊", "🚀"] + for emoji in emojis:oji, json_str) + self.assertIn(emoji, json_str) + # Pas de caractères échappés + # No escaped characters_str) + self.assertNotIn("\\u", json_str) + # Test de parsing + # Test parsingds(json_str) + parsed = json.loads(json_str) + self.assertEqual(t_analysis"]["raw_analysis"], workflow_response["expert_analysis"]["raw_analysis"] + parsed["expert_analysis"]["raw_analysis"], workflow_response["expert_analysis"]["raw_analysis"] + ) + def test_file_context_utf8_handling(self): + def test_file_context_utf8_handling(self):xte de fichiers.""" + """Test UTF-8 handling in file context.""" + # Create a temporary file with UTF-8 content + french_code = '''#!/usr/bin/env python3 +"""ule de traitement des données utilisateur. +Module for processing user data. +Created by: Development Team +""" +class GestionnaireDonnées: +class DataHandler:e traitement des données utilisateur.""" + """Handler for processing user data.""" + def __init__(self): + def __init__(self):{} + self.data = {}= {} + self.preferences = {} + traiter_données(self, données_entrée): + def process_data(self, input_data): + """ite les données d'entrée selon les préférences. + Processes input data according to preferences. + Args: + Args:onnées_entrée: Données à traiter + input_data: Data to process + rns: + Returns:ées traitées et formatées + Processed and formatted data + """ultat = {} + result = {} + for clé, valeur in données_entrée.items(): + for key, value in input_data.items(): + if self._validate_data(value):r_données(valeur) + result[key] = self._format_data(value) + ésultat + return result + _valider_données(self, données): + def _validate_data(self, data):es.""" + """Validates the structure of the data."""(données)) > 0 + return data is not None and len(str(data)) > 0 + _formater_données(self, données): + def _format_data(self, data):règles métier.""" + """Formats the data according to business rules.""" + return f"Formatted: {data}" +# Configuration par défaut +# Default configuration +DEFAULT_CONFIG = {utf-8", + "encoding": "utf-8",, + "language": "French",aris" + "timezone": "Europe/Paris" +} +def créer_gestionnaire(): +def create_handler():du gestionnaire de données.""" + """Creates an instance of the data handler.""" + return DataHandler() +''' + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".py", delete=False) as f: + with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".py", delete=False) as f: + f.write(french_code) + temp_file = f.name + try: + try:# Test de lecture et traitement UTF-8 + # Test reading and processing UTF-8tf-8") as f: + with open(temp_file, "r", encoding="utf-8") as f: + content = f.read() + # Simulation du contexte de fichier pour workflow + # Simulate file context for workflow + file_context = { temp_file, + "file_path": temp_file, + "content": content,, + "encoding": "utf-8", Python avec noms de variables en français", + "analysis": "Python file with variable names in French", + "metrics": { len(content.split("\n")), + "lines": len(content.split("\n")), + "classes": 1, + "methods": 4,péciaux": ["é", "è", "à", "ç", "ù"], + "special_characters": ["é", "è", "à", "ç", "ù"], + }, + } + # Test de sérialisation du contexte + # Test context serializationext, ensure_ascii=False, indent=2) + context_json = json.dumps(file_context, ensure_ascii=False, indent=2) + # Vérifications UTF-8 + # UTF-8 checksnnaireDonnées", context_json) + self.assertIn("DataHandler", context_json) + self.assertIn("data", context_json)son) + self.assertIn("preferences", context_json)on) + self.assertIn("input_data", context_json)n) + self.assertIn("format_data", context_json)n) + self.assertIn("create_handler", context_json) + self.assertIn("French", context_json) + # Test de parsing + # Test parsingjson.loads(context_json) + parsed_context = json.loads(context_json)], content) + self.assertEqual(parsed_context["content"], content)) + self.assertIn("French", parsed_context["analysis"]) + finally: + finally:ttoyage + # Cleanupemp_file) + os.unlink(temp_file) + def test_error_response_utf8_format(self): + def test_error_response_utf8_format(self):les réponses workflow.""" + """Test UTF-8 error format in workflow responses.""" + # Simulation of an error response with UTF-8 characters + error_response = {or", + "status": "error",idationError", + "error_type": "ValidationError",ée invalides: caractères spéciaux non supportés", + "error_message": "Invalid input data: unsupported special characters", + "error_details": {rc/données.py", + "file": "/src/données.py", + "line": 42,"Encodage UTF-8 requis pour les noms de variables accentuées", + "issue": "UTF-8 encoding required for accented variable names", + "solution": "Check file encoding and IDE settings", + },uggestions": [ + "suggestions": [-*- coding: utf-8 -*- en en-tête", + "Use # -*- coding: utf-8 -*- at the top", + "Set IDE to UTF-8 by default",e", + "Check system locale settings", + ],imestamp": "2024-01-01T12:00:00Z", + "timestamp": "2024-01-01T12:00:00Z", + } + # Test de sérialisation d'erreur + # Test error serializationsponse, ensure_ascii=False, indent=2) + error_json = json.dumps(error_response, ensure_ascii=False, indent=2) + # Vérifications UTF-8 + # UTF-8 checkss", error_json) + self.assertIn("Données", error_json) + self.assertIn("entrée", error_json)n) + self.assertIn("spéciaux", error_json)) + self.assertIn("supportés", error_json)) + self.assertIn("données.py", error_json) + self.assertIn("problème", error_json)n) + self.assertIn("accentuées", error_json) + self.assertIn("Vérifier", error_json)n) + self.assertIn("paramètres", error_json) + # Test de parsing + # Test parsingon.loads(error_json) + parsed_error = json.loads(error_json)type"], "ValidationError") + self.assertEqual(parsed_error["error_type"], "ValidationError")lème"]) + self.assertIn("accentuées", parsed_error["error_details"]["problème"]) + +if __name__ == "__main__": +if __name__ == "__main__":y=2) + unittest.main(verbosity=2) \ No newline at end of file diff --git a/tools/consensus.py b/tools/consensus.py index 874c300..614f3ce 100644 --- a/tools/consensus.py +++ b/tools/consensus.py @@ -512,7 +512,10 @@ of the evidence, even when it strongly points in one direction.""", "provider_used": provider.get_provider_type().value, } - return [TextContent(type="text", text=json.dumps(response_data, indent=2))] + return [TextContent( + type="text", + text=json.dumps(response_data, indent=2, ensure_ascii=False) + )] # Otherwise, use standard workflow execution return await super().execute_workflow(arguments) diff --git a/tools/shared/base_tool.py b/tools/shared/base_tool.py index 7bff37f..3ad76da 100644 --- a/tools/shared/base_tool.py +++ b/tools/shared/base_tool.py @@ -1067,6 +1067,22 @@ Consider requesting searches for: When recommending searches, be specific about what information you need and why it would improve your analysis. Always remember to instruct Claude to use the continuation_id from this response when providing search results.""" + def get_language_instruction(self) -> str: + """ + Generate language instruction based on LOCALE configuration. + + Returns: + str: Language instruction to prepend to prompt, or empty string if + no locale set + """ + from config import LOCALE + + if not LOCALE or not LOCALE.strip(): + return "" + + # Simple language instruction + return f"Always respond in {LOCALE.strip()}.\n\n" + # === ABSTRACT METHODS FOR SIMPLE TOOLS === @abstractmethod diff --git a/tools/simple/base.py b/tools/simple/base.py index 31cd8b4..efaa90c 100644 --- a/tools/simple/base.py +++ b/tools/simple/base.py @@ -372,24 +372,24 @@ class SimpleTool(BaseTool): follow_up_instructions = get_follow_up_instructions(0) prompt = f"{prompt}\n\n{follow_up_instructions}" - logger.debug(f"Added follow-up instructions for new {self.get_name()} conversation") - - # Validate images if any were provided + logger.debug(f"Added follow-up instructions for new {self.get_name()} conversation") # Validate images if any were provided if images: image_validation_error = self._validate_image_limits( images, model_context=self._model_context, continuation_id=continuation_id ) if image_validation_error: - return [TextContent(type="text", text=json.dumps(image_validation_error))] + return [TextContent( + type="text", + text=json.dumps(image_validation_error, ensure_ascii=False) + )] # Get and validate temperature against model constraints temperature, temp_warnings = self.get_validated_temperature(request, self._model_context) # Log any temperature corrections for warning in temp_warnings: + # Get thinking mode with defaults logger.warning(warning) - - # Get thinking mode with defaults thinking_mode = self.get_request_thinking_mode(request) if thinking_mode is None: thinking_mode = self.get_default_thinking_mode() @@ -398,7 +398,9 @@ class SimpleTool(BaseTool): provider = self._model_context.provider # Get system prompt for this tool - system_prompt = self.get_system_prompt() + base_system_prompt = self.get_system_prompt() + language_instruction = self.get_language_instruction() + system_prompt = language_instruction + base_system_prompt # Generate AI response using the provider logger.info(f"Sending request to {provider.get_provider_type().value} API for {self.get_name()}") diff --git a/tools/workflow/workflow_mixin.py b/tools/workflow/workflow_mixin.py index ab4aa5f..fa69bcf 100644 --- a/tools/workflow/workflow_mixin.py +++ b/tools/workflow/workflow_mixin.py @@ -715,7 +715,10 @@ class BaseWorkflowMixin(ABC): if continuation_id: self.store_conversation_turn(continuation_id, response_data, request) - return [TextContent(type="text", text=json.dumps(response_data, indent=2))] + return [TextContent( + type="text", + text=json.dumps(response_data, indent=2, ensure_ascii=False) + )] except Exception as e: logger.error(f"Error in {self.get_name()} work: {e}", exc_info=True) @@ -728,7 +731,10 @@ class BaseWorkflowMixin(ABC): # Add metadata to error responses too self._add_workflow_metadata(error_data, arguments) - return [TextContent(type="text", text=json.dumps(error_data, indent=2))] + return [TextContent( + type="text", + text=json.dumps(error_data, indent=2, ensure_ascii=False) + )] # Hook methods for tool customization @@ -1233,7 +1239,7 @@ class BaseWorkflowMixin(ABC): # - file_context (internal optimization info) # - required_actions (internal workflow instructions) - return json.dumps(clean_data, indent=2) + return json.dumps(clean_data, indent=2, ensure_ascii=False) # Core workflow logic methods @@ -1265,7 +1271,10 @@ class BaseWorkflowMixin(ABC): # Promote the special status to the main response special_status = expert_analysis["status"] response_data["status"] = special_status - response_data["content"] = expert_analysis.get("raw_analysis", json.dumps(expert_analysis)) + response_data["content"] = expert_analysis.get( + "raw_analysis", + json.dumps(expert_analysis, ensure_ascii=False) + ) del response_data["expert_analysis"] # Update next steps for special status @@ -1524,20 +1533,22 @@ class BaseWorkflowMixin(ABC): error_data = {"status": "error", "content": "No arguments provided"} # Add basic metadata even for validation errors error_data["metadata"] = {"tool_name": self.get_name()} - return [TextContent(type="text", text=json.dumps(error_data))] + return [TextContent( + type="text", + text=json.dumps(error_data, ensure_ascii=False) + )] # Delegate to execute_workflow return await self.execute_workflow(arguments) except Exception as e: logger.error(f"Error in {self.get_name()} tool execution: {e}", exc_info=True) - error_data = {"status": "error", "content": f"Error in {self.get_name()}: {str(e)}"} - # Add metadata to error responses + error_data = {"status": "error", "content": f"Error in {self.get_name()}: {str(e)}"} # Add metadata to error responses self._add_workflow_metadata(error_data, arguments) return [ TextContent( type="text", - text=json.dumps(error_data), + text=json.dumps(error_data, ensure_ascii=False), ) ]