OpenAI MCP Server

  • claude_code
#!/usr/bin/env python3 # claude.py """Claude Code Python Edition - CLI entry point.""" import os import sys import logging import argparse from typing import Dict, List, Optional, Any import json import signal from datetime import datetime import typer from rich.console import Console from rich.panel import Panel from rich.markdown import Markdown from rich.prompt import Prompt from rich.syntax import Syntax from rich.logging import RichHandler from dotenv import load_dotenv from claude_code.lib.providers import get_provider, list_available_providers from claude_code.lib.tools.base import ToolRegistry from claude_code.lib.tools.manager import ToolExecutionManager from claude_code.lib.tools.file_tools import register_file_tools from claude_code.lib.ui.tool_visualizer import ToolCallVisualizer, MultiPanelLayout from claude_code.lib.monitoring.cost_tracker import CostTracker # Configure logging logging.basicConfig( level=logging.INFO, format="%(message)s", datefmt="[%X]", handlers=[RichHandler(rich_tracebacks=True)] ) logger = logging.getLogger("claude_code") # Load environment variables load_dotenv() # Get version from package VERSION = "0.1.0" # Create typer app app = typer.Typer(help="Claude Code Python Edition") console = Console() # Global state conversation: List[Dict[str, Any]] = [] tool_registry = ToolRegistry() tool_manager: Optional[ToolExecutionManager] = None cost_tracker: Optional[CostTracker] = None visualizer: Optional[ToolCallVisualizer] = None provider_name: str = "" model_name: str = "" user_config: Dict[str, Any] = {} def initialize_tools() -> None: """Initialize all available tools.""" global tool_registry, tool_manager # Create the registry and manager tool_registry = ToolRegistry() tool_manager = ToolExecutionManager(tool_registry) # Register file tools register_file_tools(tool_registry) # TODO: Register more tools # register_search_tools(tool_registry) # register_bash_tools(tool_registry) # register_agent_tools(tool_registry) logger.info(f"Initialized {len(tool_registry.get_all_tools())} tools") def setup_visualizer() -> None: """Set up the tool visualizer with callbacks.""" global tool_manager, visualizer if not tool_manager: return # Create visualizer visualizer = ToolCallVisualizer(console) # Set up callbacks def progress_callback(tool_call_id: str, progress: float) -> None: if visualizer: visualizer.update_progress(tool_call_id, progress) def result_callback(tool_call_id: str, result: Any) -> None: if visualizer: visualizer.complete_tool_call(tool_call_id, result) tool_manager.set_progress_callback(progress_callback) tool_manager.set_result_callback(result_callback) def load_configuration() -> Dict[str, Any]: """Load user configuration from file.""" config_path = os.path.expanduser("~/.config/claude_code/config.json") # Default configuration default_config = { "provider": "openai", "model": None, # Use provider default "budget_limit": None, "history_file": os.path.expanduser("~/.config/claude_code/usage_history.json"), "ui": { "theme": "dark", "show_tool_calls": True, "show_cost": True } } # If configuration file doesn't exist, create it with defaults if not os.path.exists(config_path): try: os.makedirs(os.path.dirname(config_path), exist_ok=True) with open(config_path, 'w', encoding='utf-8') as f: json.dump(default_config, f, indent=2) except Exception as e: logger.warning(f"Failed to create default configuration: {e}") return default_config # Load configuration try: with open(config_path, 'r', encoding='utf-8') as f: config = json.load(f) # Merge with defaults for any missing keys for key, value in default_config.items(): if key not in config: config[key] = value return config except Exception as e: logger.warning(f"Failed to load configuration: {e}") return default_config def handle_compact_command() -> str: """Handle the /compact command to compress conversation history.""" global conversation, provider_name, model_name if not conversation: return "No conversation to compact." # Add a system message requesting summarization compact_prompt = ( "Summarize the conversation so far, focusing on the key points, decisions, and context. " "Keep important details about the code and tasks. Retain critical file paths, commands, " "and code snippets. The summary should be concise but complete enough to continue the " "conversation effectively." ) conversation.append({"role": "user", "content": compact_prompt}) # Get the provider provider = get_provider(provider_name, model=model_name) # Make non-streaming API call for compaction response = provider.generate_completion(conversation, stream=False) # Extract summary summary = response["content"] or "" # Reset conversation with summary system_message = next((m for m in conversation if m["role"] == "system"), None) if system_message: conversation = [system_message] else: conversation = [] # Add compacted context conversation.append({ "role": "system", "content": f"This is a compacted conversation. Previous context: {summary}" }) return "Conversation compacted successfully." def handle_help_command() -> str: """Handle the /help command.""" help_text = """ # Claude Code Python Edition Help ## Commands - **/help**: Show this help message - **/compact**: Compact the conversation to reduce token usage - **/version**: Show version information - **/providers**: List available LLM providers - **/cost**: Show cost and usage information - **/budget [amount]**: Set a budget limit (e.g., /budget 5.00) - **/quit, /exit**: Exit the application ## Routine Commands - **/routine list**: List all available routines - **/routine create <name> <description>**: Create a routine from recent tool executions - **/routine run <name>**: Run a routine - **/routine delete <name>**: Delete a routine ## Tools Claude Code has access to these tools: - **View**: Read files - **Edit**: Edit files (replace text) - **Replace**: Overwrite or create files - **GlobTool**: Find files by pattern - **GrepTool**: Search file contents - **LS**: List directory contents - **Bash**: Execute shell commands ## CLI Commands - **claude**: Start the Claude Code assistant (main interface) - **claude mcp-client**: Start the MCP client to connect to MCP servers - Usage: `claude mcp-client path/to/server.py [--model MODEL]` - **claude mcp-multi-agent**: Start the multi-agent MCP client with synchronized agents - Usage: `claude mcp-multi-agent path/to/server.py [--config CONFIG_FILE]` ## Multi-Agent Commands When using the multi-agent client: - **/agents**: List all active agents - **/talk <agent> <message>**: Send a direct message to a specific agent - **/history**: Show message history - **/help**: Show multi-agent help ## Tips - Be specific about file paths when requesting file operations - For complex tasks, break them down into smaller steps - Use /compact periodically for long sessions to save tokens - Create routines for repetitive sequences of tool operations - In multi-agent mode, use agent specialization for complex problems """ return help_text def handle_version_command() -> str: """Handle the /version command.""" import platform python_version = platform.python_version() version_info = f""" # Claude Code Python Edition v{VERSION} - Python: {python_version} - Provider: {provider_name} - Model: {model_name} - Tools: {len(tool_registry.get_all_tools()) if tool_registry else 0} available """ return version_info def handle_providers_command() -> str: """Handle the /providers command.""" providers = list_available_providers() providers_text = "# Available LLM Providers\n\n" for name, info in providers.items(): providers_text += f"## {info['name']}\n" if info['available']: providers_text += f"- Status: Available\n" providers_text += f"- Current model: {info['current_model']}\n" providers_text += f"- Available models: {', '.join(info['models'])}\n" else: providers_text += f"- Status: Not available ({info['error']})\n" providers_text += "\n" return providers_text def handle_cost_command() -> str: """Handle the /cost command.""" global cost_tracker if not cost_tracker: return "Cost tracking is not available." # Generate a usage report return cost_tracker.generate_usage_report(format="markdown") def handle_budget_command(args: List[str]) -> str: """Handle the /budget command.""" global cost_tracker if not cost_tracker: return "Cost tracking is not available." if not args: # Show current budget budget = cost_tracker.check_budget() if not budget["has_budget"]: return "No budget limit is currently set." return f"Current budget: ${budget['limit']:.2f} (${budget['used']:.2f} used, ${budget['remaining']:.2f} remaining)" # Set new budget try: budget_amount = float(args[0]) if budget_amount <= 0: return "Budget must be a positive number." cost_tracker.budget_limit = budget_amount # Update configuration user_config["budget_limit"] = budget_amount # Save configuration config_path = os.path.expanduser("~/.config/claude_code/config.json") try: with open(config_path, 'w', encoding='utf-8') as f: json.dump(user_config, f, indent=2) except Exception as e: logger.warning(f"Failed to save configuration: {e}") return f"Budget set to ${budget_amount:.2f}" except ValueError: return f"Invalid budget amount: {args[0]}" def handle_routine_list_command() -> str: """Handle the /routine list command.""" global tool_manager if not tool_manager: return "Tool manager is not initialized." routines = tool_manager.registry.get_all_routines() if not routines: return "No routines available." routines_text = "# Available Routines\n\n" for routine in routines: usage = f" (Used {routine.usage_count} times)" if routine.usage_count > 0 else "" last_used = "" if routine.last_used_at: last_used_time = datetime.fromtimestamp(routine.last_used_at) last_used = f" (Last used: {last_used_time.strftime('%Y-%m-%d %H:%M')})" routines_text += f"## {routine.name}{usage}{last_used}\n" routines_text += f"{routine.description}\n\n" routines_text += f"**Steps:** {len(routine.steps)}\n\n" return routines_text def handle_routine_create_command(args: List[str]) -> str: """Handle the /routine create command.""" global tool_manager, visualizer if not tool_manager: return "Tool manager is not initialized." if len(args) < 2: return "Usage: /routine create <name> <description>" name = args[0] description = " ".join(args[1:]) # Get recent tool results from visualizer if not visualizer or not hasattr(visualizer, "recent_tool_results"): return "No recent tool executions to create a routine from." recent_tool_results = visualizer.recent_tool_results if not recent_tool_results: return "No recent tool executions to create a routine from." try: routine_id = tool_manager.create_routine_from_tool_history( name, description, recent_tool_results ) return f"Created routine '{name}' with {len(recent_tool_results)} steps." except Exception as e: logger.exception(f"Error creating routine: {e}") return f"Error creating routine: {str(e)}" def handle_routine_run_command(args: List[str]) -> str: """Handle the /routine run command.""" global tool_manager, visualizer if not tool_manager: return "Tool manager is not initialized." if not args: return "Usage: /routine run <name>" name = args[0] # Check if routine exists routine = tool_manager.registry.get_routine(name) if not routine: return f"Routine '{name}' not found." try: # Execute the routine execution_id = tool_manager.execute_routine(name) # Wait for completion while True: routine_status = tool_manager.routine_manager.get_active_routines().get(execution_id, {}) if routine_status.get("status") != "running": break time.sleep(0.1) # Get results results = tool_manager.get_routine_results(execution_id) if not results: return f"Routine '{name}' completed but returned no results." # Format results result_text = f"# Routine '{name}' Results\n\n" result_text += f"Executed {len(results)} steps:\n\n" for i, result in enumerate(results): status = "✅" if result.status == "success" else "❌" result_text += f"## Step {i+1}: {result.name} {status}\n" result_text += f"```\n{result.result}\n```\n\n" return result_text except Exception as e: logger.exception(f"Error executing routine: {e}") return f"Error executing routine: {str(e)}" def handle_routine_delete_command(args: List[str]) -> str: """Handle the /routine delete command.""" global tool_manager if not tool_manager: return "Tool manager is not initialized." if not args: return "Usage: /routine delete <name>" name = args[0] # Check if routine exists routine = tool_manager.registry.get_routine(name) if not routine: return f"Routine '{name}' not found." try: # Remove from registry and save tool_manager.registry.routines.pop(name, None) tool_manager.registry._save_routines() return f"Deleted routine '{name}'." except Exception as e: logger.exception(f"Error deleting routine: {e}") return f"Error deleting routine: {str(e)}" def process_special_command(user_input: str) -> Optional[str]: """Process special commands starting with /.""" # Split into command and arguments parts = user_input.strip().split() command = parts[0].lower() args = parts[1:] # Handle commands if command == "/help": return handle_help_command() elif command == "/compact": return handle_compact_command() elif command == "/version": return handle_version_command() elif command == "/providers": return handle_providers_command() elif command == "/cost": return handle_cost_command() elif command == "/budget": return handle_budget_command(args) elif command in ["/quit", "/exit"]: console.print("[bold yellow]Goodbye![/bold yellow]") sys.exit(0) # Handle routine commands elif command == "/routine": if not args: return "Usage: /routine [list|create|run|delete]" subcmd = args[0].lower() if subcmd == "list": return handle_routine_list_command() elif subcmd == "create": return handle_routine_create_command(args[1:]) elif subcmd == "run": return handle_routine_run_command(args[1:]) elif subcmd == "delete": return handle_routine_delete_command(args[1:]) else: return f"Unknown routine command: {subcmd}\nUsage: /routine [list|create|run|delete]" # Not a recognized command return None def process_tool_calls(tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Process tool calls and return results. Args: tool_calls: List of tool call dictionaries Returns: List of tool responses """ global tool_manager, visualizer if not tool_manager: logger.error("Tool manager not initialized") return [] # Add tool calls to visualizer if visualizer: for tool_call in tool_calls: function_name = tool_call.get("function", {}).get("name", "") tool_call_id = tool_call.get("id", "unknown") arguments_str = tool_call.get("function", {}).get("arguments", "{}") try: parameters = json.loads(arguments_str) visualizer.add_tool_call(tool_call_id, function_name, parameters) except json.JSONDecodeError: visualizer.add_tool_call(tool_call_id, function_name, {}) # Execute tools in parallel tool_results = tool_manager.execute_tools_parallel(tool_calls) # Format results for the conversation tool_responses = [] for result in tool_results: tool_responses.append({ "tool_call_id": result.tool_call_id, "role": "tool", "name": result.name, "content": result.result }) return tool_responses @app.command(name="mcp-client") def mcp_client( server_script: str = typer.Argument(..., help="Path to the server script (.py or .js)"), model: str = typer.Option("claude-3-5-sonnet-20241022", "--model", "-m", help="Claude model to use") ): """Run the MCP client to interact with an MCP server.""" from claude_code.commands.client import execute as client_execute import argparse # Create a namespace with the arguments args = argparse.Namespace() args.server_script = server_script args.model = model # Execute the client return client_execute(args) @app.command(name="mcp-multi-agent") def mcp_multi_agent( server_script: str = typer.Argument(..., help="Path to the server script (.py or .js)"), config: str = typer.Option(None, "--config", "-c", help="Path to agent configuration JSON file") ): """Run the multi-agent MCP client with agent synchronization.""" from claude_code.commands.multi_agent_client import execute as multi_agent_execute import argparse # Create a namespace with the arguments args = argparse.Namespace() args.server_script = server_script args.config = config # Execute the multi-agent client return multi_agent_execute(args) @app.command() def main( provider: str = typer.Option(None, "--provider", "-p", help="LLM provider to use"), model: str = typer.Option(None, "--model", "-m", help="Model to use"), budget: Optional[float] = typer.Option(None, "--budget", "-b", help="Budget limit in dollars"), system_prompt: Optional[str] = typer.Option(None, "--system", "-s", help="System prompt file"), verbose: bool = typer.Option(False, "--verbose", "-v", help="Enable verbose output") ): """Claude Code Python Edition - A LLM-powered coding assistant.""" global conversation, tool_registry, tool_manager, cost_tracker, visualizer global provider_name, model_name, user_config # Set logging level if verbose: logging.getLogger("claude_code").setLevel(logging.DEBUG) # Show welcome message console.print(Panel.fit( f"[bold green]Claude Code Python Edition v{VERSION}[/bold green]\n" "Type your questions or commands. Use /help for available commands.", title="Welcome", border_style="green" )) # Load configuration user_config = load_configuration() # Override with command line arguments if provider: user_config["provider"] = provider if model: user_config["model"] = model if budget is not None: user_config["budget_limit"] = budget # Set provider and model provider_name = user_config["provider"] model_name = user_config["model"] try: # Initialize tools initialize_tools() # Set up cost tracking cost_tracker = CostTracker( budget_limit=user_config["budget_limit"], history_file=user_config["history_file"] ) # Get provider provider = get_provider(provider_name, model=model_name) provider_name = provider.name model_name = provider.current_model logger.info(f"Using {provider_name} with model {model_name}") # Set up tool visualizer setup_visualizer() if visualizer: visualizer.start() # Load system prompt system_message = "" if system_prompt: try: with open(system_prompt, 'r', encoding='utf-8') as f: system_message = f.read() except Exception as e: logger.error(f"Failed to load system prompt: {e}") system_message = get_default_system_prompt() else: system_message = get_default_system_prompt() # Initialize conversation conversation = [{"role": "system", "content": system_message}] # Main interaction loop while True: try: # Get user input user_input = Prompt.ask("\n[bold blue]>>[/bold blue]") # Handle special commands if user_input.startswith("/"): result = process_special_command(user_input) if result: console.print(Markdown(result)) continue # Add user message to conversation conversation.append({"role": "user", "content": user_input}) # Get schemas for all tools tool_schemas = tool_registry.get_tool_schemas() if tool_registry else None # Call the LLM with console.status("[bold blue]Thinking...[/bold blue]", spinner="dots"): # Stream the response response_stream = provider.generate_completion( messages=conversation, tools=tool_schemas, stream=True ) # Track tool calls from streaming response current_content = "" current_tool_calls = [] # Process streaming response for chunk in response_stream: # If there's content, print it if chunk.get("content"): content_piece = chunk["content"] current_content += content_piece console.print(content_piece, end="") # Process tool calls if chunk.get("tool_calls") and not chunk.get("delta", True): # This is a complete tool call current_tool_calls = chunk["tool_calls"] break console.print() # Add newline after content # Add assistant response to conversation conversation.append({ "role": "assistant", "content": current_content, "tool_calls": current_tool_calls }) # Process tool calls if any if current_tool_calls: console.print("[bold green]Executing tools...[/bold green]") # Process tool calls tool_responses = process_tool_calls(current_tool_calls) # Add tool responses to conversation conversation.extend(tool_responses) # Continue the conversation with tool responses console.print("[bold blue]Continuing with tool results...[/bold blue]") follow_up = provider.generate_completion( messages=conversation, tools=tool_schemas, stream=False ) follow_up_text = follow_up.get("content", "") if follow_up_text: console.print(Markdown(follow_up_text)) # Add to conversation conversation.append({ "role": "assistant", "content": follow_up_text }) # Track token usage and cost if cost_tracker: # Get token counts - this is an approximation token_counts = provider.count_message_tokens(conversation[-3:]) cost_info = provider.cost_per_1k_tokens # Add request to tracker cost_tracker.add_request( provider=provider_name, model=model_name, tokens_input=token_counts["input"], tokens_output=token_counts.get("output", 0) or 150, # Estimate if not available input_cost_per_1k=cost_info["input"], output_cost_per_1k=cost_info["output"] ) # Check budget budget_status = cost_tracker.check_budget() if budget_status["has_budget"] and budget_status["status"] in ["critical", "exceeded"]: console.print(f"[bold red]{budget_status['message']}[/bold red]") except KeyboardInterrupt: console.print("\n[bold yellow]Operation cancelled by user.[/bold yellow]") continue except Exception as e: logger.exception(f"Error: {str(e)}") console.print(f"[bold red]Error:[/bold red] {str(e)}") finally: # Clean up if visualizer: visualizer.stop() # Save cost history if cost_tracker and hasattr(cost_tracker, '_save_history'): cost_tracker._save_history() def get_default_system_prompt() -> str: """Get the default system prompt.""" return """You are Claude Code Python Edition, a CLI tool that helps users with software engineering tasks. Use the available tools to assist the user with their requests. # Tone and style You should be concise, direct, and to the point. When you run a non-trivial bash command, you should explain what the command does and why you are running it. Output text to communicate with the user; all text you output outside of tool use is displayed to the user. Remember that your output will be displayed on a command line interface. # Tool usage policy - When doing file search, remember to search effectively with the available tools. - Always use the appropriate tool for the task. - Use parallel tool calls when appropriate to improve performance. - NEVER commit changes unless the user explicitly asks you to. # Routines You have access to Routines, which are sequences of tool calls that can be created and reused. To create a routine from recent tool executions, use `/routine create <name> <description>`. To run a routine, use `/routine run <name>`. Routines are ideal for repetitive task sequences like: - Deep research across multiple sources - Multi-step code updates across files - Complex search and replace operations - Data processing pipelines # Tasks The user will primarily request you perform software engineering tasks: 1. Solving bugs 2. Adding new functionality 3. Refactoring code 4. Explaining code 5. Writing tests For these tasks: 1. Use search tools to understand the codebase 2. Implement solutions using the available tools 3. Verify solutions with tests if possible 4. Run lint and typecheck commands when appropriate 5. Consider creating routines for repetitive operations # Code style - Follow the existing code style of the project - Maintain consistent naming conventions - Use appropriate libraries that are already in the project - Add comments when code is complex or non-obvious IMPORTANT: You should minimize output tokens as much as possible while maintaining helpfulness, quality, and accuracy. Answer concisely with short lines of text unless the user asks for detail. """ if __name__ == "__main__": # Handle Ctrl+C gracefully signal.signal(signal.SIGINT, lambda sig, frame: sys.exit(0)) # Run app app()