"""
Galaxy Brain MCP Server
=======================
Think. Do. Done.
A complete cognitive loop MCP server that combines sequential thinking
(planning with revision) with sequential doing (execution with variable piping).
Usage:
python -m galaxy_brain.server
uvx galaxy-brain
"""
import asyncio
import json
from typing import Any, Sequence
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
EmbeddedResource,
)
from .config import get_config, set_config, GalaxyBrainConfig
from .logger import setup_logger, get_logger
from .services import ThinkingService, DoingService, BridgeService
# Initialize server
server = Server("galaxy-brain")
# Services (initialized lazily)
_thinking: ThinkingService = None
_doing: DoingService = None
_bridge: BridgeService = None
def _get_services():
"""Lazy initialization of services"""
global _thinking, _doing, _bridge
if _thinking is None:
_thinking = ThinkingService()
if _doing is None:
_doing = DoingService()
if _bridge is None:
_bridge = BridgeService(_thinking, _doing)
return _thinking, _doing, _bridge
# =============================================================================
# Tool Definitions
# =============================================================================
TOOLS = [
# Thinking tools
Tool(
name="start_thinking",
description=(
"Start a new thinking session to reason through a problem step by step. "
"Returns a session_id to use with other thinking tools."
),
inputSchema={
"type": "object",
"properties": {
"problem": {
"type": "string",
"description": "The problem or question to think through"
},
"initial_estimate": {
"type": "integer",
"description": "Initial estimate of thoughts needed (default: 5)",
"default": 5
}
},
"required": ["problem"]
}
),
Tool(
name="think",
description=(
"Add a thought to a thinking session. Use this to reason through "
"a problem step by step. Set next_thought_needed=false when done."
),
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "The thinking session ID"
},
"thought": {
"type": "string",
"description": "The thought content"
},
"confidence": {
"type": "number",
"description": "Confidence in this thought (0-1)",
"minimum": 0,
"maximum": 1
},
"next_thought_needed": {
"type": "boolean",
"description": "Whether more thinking is needed (default: true)",
"default": True
}
},
"required": ["session_id", "thought"]
}
),
Tool(
name="revise",
description=(
"Revise a previous thought in a thinking session. Use when you realize "
"an earlier thought was wrong or incomplete."
),
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "The thinking session ID"
},
"revises_thought": {
"type": "integer",
"description": "The thought number to revise"
},
"revised_content": {
"type": "string",
"description": "The revised thought content"
},
"reason": {
"type": "string",
"description": "Reason for the revision"
}
},
"required": ["session_id", "revises_thought", "revised_content"]
}
),
Tool(
name="branch",
description=(
"Create a new branch to explore an alternative approach. "
"Useful when you want to try a different direction without "
"losing your current train of thought."
),
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "The thinking session ID"
},
"branch_from": {
"type": "integer",
"description": "The thought number to branch from"
},
"branch_name": {
"type": "string",
"description": "Optional name for the branch"
},
"first_thought": {
"type": "string",
"description": "Optional first thought in the new branch"
}
},
"required": ["session_id", "branch_from"]
}
),
Tool(
name="conclude",
description=(
"Conclude a thinking session with a final synthesis. "
"Call this when you've finished reasoning through the problem."
),
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "The thinking session ID"
},
"conclusion": {
"type": "string",
"description": "The final conclusion/synthesis"
},
"confidence": {
"type": "number",
"description": "Overall confidence in the conclusion (0-1)"
}
},
"required": ["session_id", "conclusion"]
}
),
# Doing tools
Tool(
name="execute_batch",
description=(
"Execute a batch of operations sequentially with variable piping. "
"Use $results[N].path.to.value to reference previous results. "
"Available services: python, shell, file, transform."
),
inputSchema={
"type": "object",
"properties": {
"operations": {
"type": "array",
"description": "List of operations to execute",
"items": {
"type": "object",
"properties": {
"service": {
"type": "string",
"enum": ["python", "shell", "file", "transform"]
},
"method": {"type": "string"},
"params": {"type": "object"}
},
"required": ["service", "method"]
}
},
"batch_name": {
"type": "string",
"description": "Optional name for the batch"
},
"stop_on_error": {
"type": "boolean",
"description": "Stop on first error (default: true)",
"default": True
}
},
"required": ["operations"]
}
),
Tool(
name="execute_single",
description="Execute a single operation (convenience wrapper for execute_batch)",
inputSchema={
"type": "object",
"properties": {
"service": {
"type": "string",
"description": "Service name",
"enum": ["python", "shell", "file", "transform"]
},
"method": {
"type": "string",
"description": "Method name"
},
"params": {
"type": "object",
"description": "Operation parameters"
}
},
"required": ["service", "method"]
}
),
Tool(
name="list_services",
description="List available services and their methods for execute_batch",
inputSchema={
"type": "object",
"properties": {}
}
),
# Bridge tools
Tool(
name="generate_plan",
description=(
"Generate an action plan from a completed thinking session. "
"Converts thoughts into executable operations."
),
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "The thinking session ID"
},
"operations": {
"type": "array",
"description": "Optional manual operations (skips auto-generation)",
"items": {"type": "object"}
}
},
"required": ["session_id"]
}
),
Tool(
name="execute_plan",
description="Execute a generated action plan",
inputSchema={
"type": "object",
"properties": {
"plan_id": {
"type": "string",
"description": "The plan ID to execute"
},
"force": {
"type": "boolean",
"description": "Execute even if validation failed",
"default": False
}
},
"required": ["plan_id"]
}
),
Tool(
name="think_and_do",
description=(
"Complete cognitive loop in one call: think through a problem, "
"then execute operations. The ultimate galaxy brain move."
),
inputSchema={
"type": "object",
"properties": {
"problem": {
"type": "string",
"description": "The problem to solve"
},
"thoughts": {
"type": "array",
"description": "Your reasoning steps",
"items": {"type": "string"}
},
"operations": {
"type": "array",
"description": "Operations to execute after thinking",
"items": {"type": "object"}
},
"execute": {
"type": "boolean",
"description": "Whether to execute immediately (default: true)",
"default": True
}
},
"required": ["problem", "thoughts", "operations"]
}
),
# Utility tools
Tool(
name="get_session",
description="Get the current state of a thinking session",
inputSchema={
"type": "object",
"properties": {
"session_id": {
"type": "string",
"description": "The session ID"
}
},
"required": ["session_id"]
}
),
Tool(
name="list_sessions",
description="List all thinking sessions",
inputSchema={
"type": "object",
"properties": {
"include_completed": {
"type": "boolean",
"description": "Include completed sessions",
"default": False
}
}
}
),
Tool(
name="get_plan",
description="Get a plan by ID",
inputSchema={
"type": "object",
"properties": {
"plan_id": {
"type": "string",
"description": "The plan ID"
}
},
"required": ["plan_id"]
}
),
Tool(
name="list_plans",
description="List all action plans",
inputSchema={
"type": "object",
"properties": {}
}
)
]
# =============================================================================
# Tool Handlers
# =============================================================================
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools"""
return TOOLS
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> Sequence[TextContent | EmbeddedResource]:
"""Handle tool calls"""
thinking, doing, bridge = _get_services()
logger = get_logger('server')
logger.info(f"Tool call: {name}")
try:
# Thinking tools
if name == "start_thinking":
result = thinking.start_thinking(
problem=arguments["problem"],
initial_estimate=arguments.get("initial_estimate", 5)
)
elif name == "think":
result = thinking.think(
session_id=arguments["session_id"],
thought=arguments["thought"],
confidence=arguments.get("confidence"),
next_thought_needed=arguments.get("next_thought_needed", True)
)
elif name == "revise":
result = thinking.revise(
session_id=arguments["session_id"],
revises_thought=arguments["revises_thought"],
revised_content=arguments["revised_content"],
reason=arguments.get("reason")
)
elif name == "branch":
result = thinking.branch(
session_id=arguments["session_id"],
branch_from=arguments["branch_from"],
branch_name=arguments.get("branch_name"),
first_thought=arguments.get("first_thought")
)
elif name == "conclude":
result = thinking.conclude(
session_id=arguments["session_id"],
conclusion=arguments["conclusion"],
confidence=arguments.get("confidence")
)
# Doing tools
elif name == "execute_batch":
result = doing.execute_batch(
operations=arguments["operations"],
batch_name=arguments.get("batch_name"),
stop_on_error=arguments.get("stop_on_error", True)
)
elif name == "execute_single":
result = doing.execute_single(
service=arguments["service"],
method=arguments["method"],
params=arguments.get("params", {})
)
elif name == "list_services":
result = doing.list_services()
# Bridge tools
elif name == "generate_plan":
result = bridge.generate_plan(
session_id=arguments["session_id"],
operations=arguments.get("operations")
)
elif name == "execute_plan":
result = bridge.execute_plan(
plan_id=arguments["plan_id"],
force=arguments.get("force", False)
)
elif name == "think_and_do":
result = bridge.think_and_do(
problem=arguments["problem"],
thoughts=arguments["thoughts"],
operations=arguments["operations"],
execute=arguments.get("execute", True)
)
# Utility tools
elif name == "get_session":
result = thinking.get_session(arguments["session_id"])
elif name == "list_sessions":
result = thinking.list_sessions(
include_completed=arguments.get("include_completed", False)
)
elif name == "get_plan":
result = bridge.get_plan(arguments["plan_id"])
elif name == "list_plans":
result = bridge.list_plans()
else:
result = {
"success": False,
"error": f"Unknown tool: {name}"
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2, default=str)
)]
except Exception as e:
logger.exception(f"Tool error: {name}")
return [TextContent(
type="text",
text=json.dumps({
"success": False,
"error": str(e),
"error_type": type(e).__name__
}, indent=2)
)]
# =============================================================================
# Server Entry Point
# =============================================================================
async def serve():
"""Run the Galaxy Brain MCP server"""
config = get_config()
logger = setup_logger(
name="galaxy-brain",
level=config.log_level,
log_file=config.log_file
)
logger.info("=" * 60)
logger.info("Galaxy Brain MCP Server")
logger.info("Think. Do. Done.")
logger.info("=" * 60)
logger.info(f"Version: {config.server_version}")
logger.info(f"Max thoughts: {config.thinking.max_thoughts}")
logger.info(f"Max operations: {config.doing.max_operations}")
logger.info("=" * 60)
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
def main():
"""Main entry point"""
asyncio.run(serve())
if __name__ == "__main__":
main()
