OpenAI MCP Server
by arthurcolle
- 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()