"""
Standardized response builders for Galaxy Brain MCP
"""
from typing import Any, Dict, List, Optional
from datetime import datetime
def success_response(
message: str,
data: Optional[Dict[str, Any]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Build a successful response
Args:
message: Success message
data: Response data
metadata: Additional metadata
Returns:
Standardized success response
"""
response = {
'success': True,
'message': message,
'timestamp': datetime.utcnow().isoformat()
}
if data:
response['data'] = data
if metadata:
response['metadata'] = metadata
return response
def error_response(
message: str,
error_type: str = "Error",
details: Optional[Dict[str, Any]] = None,
suggestion: Optional[str] = None
) -> Dict[str, Any]:
"""
Build an error response
Args:
message: Error message
error_type: Type of error
details: Error details
suggestion: Suggested fix
Returns:
Standardized error response
"""
response = {
'success': False,
'error': error_type,
'message': message,
'timestamp': datetime.utcnow().isoformat()
}
if details:
response['details'] = details
if suggestion:
response['suggestion'] = suggestion
return response
def thinking_response(
thought_number: int,
total_thoughts: int,
thought: str,
next_thought_needed: bool,
is_revision: bool = False,
revises_thought: Optional[int] = None,
branch_id: Optional[str] = None,
branch_from: Optional[int] = None,
confidence: Optional[float] = None,
metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Build a thinking/planning response
Args:
thought_number: Current thought number
total_thoughts: Estimated total thoughts
thought: The actual thought content
next_thought_needed: Whether more thinking is needed
is_revision: Whether this revises a previous thought
revises_thought: Which thought is being revised
branch_id: Branch identifier if branching
branch_from: Thought number this branches from
confidence: Confidence level (0-1)
metadata: Additional metadata
Returns:
Standardized thinking response
"""
response = {
'success': True,
'phase': 'thinking',
'thought_number': thought_number,
'total_thoughts': total_thoughts,
'thought': thought,
'next_thought_needed': next_thought_needed,
'timestamp': datetime.utcnow().isoformat()
}
if is_revision:
response['is_revision'] = True
response['revises_thought'] = revises_thought
if branch_id:
response['branch_id'] = branch_id
response['branch_from'] = branch_from
if confidence is not None:
response['confidence'] = confidence
if metadata:
response['metadata'] = metadata
return response
def doing_response(
operation: str,
operation_index: int,
total_operations: int,
success: bool,
result: Any = None,
error: Optional[str] = None,
duration_seconds: Optional[float] = None,
variables_used: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Build an execution/doing response
Args:
operation: Operation description
operation_index: Current operation index
total_operations: Total operations in batch
success: Whether operation succeeded
result: Operation result
error: Error message if failed
duration_seconds: Execution duration
variables_used: Variables substituted from previous results
Returns:
Standardized doing response
"""
response = {
'success': success,
'phase': 'doing',
'operation': operation,
'operation_index': operation_index,
'total_operations': total_operations,
'timestamp': datetime.utcnow().isoformat()
}
if result is not None:
response['result'] = result
if error:
response['error'] = error
if duration_seconds is not None:
response['duration_seconds'] = round(duration_seconds, 3)
if variables_used:
response['variables_used'] = variables_used
return response
def bridge_response(
thoughts_count: int,
operations_generated: int,
operations: List[Dict[str, Any]],
ready_to_execute: bool,
validation_passed: bool = True,
validation_errors: Optional[List[str]] = None,
estimated_duration: Optional[float] = None
) -> Dict[str, Any]:
"""
Build a bridge (thought-to-action) response
Args:
thoughts_count: Number of thoughts processed
operations_generated: Number of operations generated
operations: The generated operations
ready_to_execute: Whether the plan is ready to execute
validation_passed: Whether validation passed
validation_errors: List of validation errors
estimated_duration: Estimated execution time
Returns:
Standardized bridge response
"""
response = {
'success': validation_passed,
'phase': 'bridge',
'thoughts_processed': thoughts_count,
'operations_generated': operations_generated,
'operations': operations,
'ready_to_execute': ready_to_execute,
'timestamp': datetime.utcnow().isoformat()
}
if not validation_passed:
response['validation_errors'] = validation_errors or []
if estimated_duration is not None:
response['estimated_duration_seconds'] = round(estimated_duration, 2)
return response
def cognitive_loop_response(
thinking_summary: Dict[str, Any],
doing_summary: Dict[str, Any],
total_thoughts: int,
total_operations: int,
operations_succeeded: int,
operations_failed: int,
total_duration: float,
final_result: Any = None
) -> Dict[str, Any]:
"""
Build a complete cognitive loop response
Args:
thinking_summary: Summary of thinking phase
doing_summary: Summary of doing phase
total_thoughts: Total thoughts generated
total_operations: Total operations executed
operations_succeeded: Successful operations count
operations_failed: Failed operations count
total_duration: Total execution time
final_result: Final result of the loop
Returns:
Complete cognitive loop response
"""
return {
'success': operations_failed == 0,
'phase': 'complete',
'thinking': thinking_summary,
'doing': doing_summary,
'summary': {
'total_thoughts': total_thoughts,
'total_operations': total_operations,
'operations_succeeded': operations_succeeded,
'operations_failed': operations_failed,
'success_rate': (
round(operations_succeeded / total_operations * 100, 1)
if total_operations > 0 else 0
),
'total_duration_seconds': round(total_duration, 3)
},
'final_result': final_result,
'timestamp': datetime.utcnow().isoformat()
}
