model_info_tools.py•15.9 kB
"""
MCP Tools for Model Information and Capabilities
Exposes model availability, costs, and capabilities to connected AI
"""
import json
from typing import Dict, List, Any, Optional
from pathlib import Path
import asyncio
from ..core.model_info_manager import ModelInfoManager, get_model_manager
from ..core.dynamic_context_manager import DynamicContextManager
class ModelInfoTools:
"""Tools for querying model information via MCP"""
def __init__(self):
self.manager = get_model_manager()
async def get_available_models(self) -> Dict[str, Any]:
"""
Get list of all available models with their capabilities
Returns:
Dictionary of provider -> models with capabilities
"""
result = {
"providers": {},
"summary": {
"total_models": 0,
"providers_available": 0,
"cheapest_model": None,
"largest_context": None
}
}
# Get available models by provider
available = self.manager.get_available_models()
for provider, model_list in available.items():
provider_info = {
"models": [],
"has_api_key": self.manager.api_keys.get(provider) is not None
}
for model_name in model_list:
# Get detailed info for each model
info = await self.manager.get_model_info(model_name, provider)
model_details = {
"name": model_name,
"context_window": info.context_window,
"max_output": info.max_output_tokens,
"usable_context": info.usable_context,
"supports_vision": info.supports_vision,
"supports_functions": info.supports_functions,
"supports_json_mode": info.supports_json_mode,
"cost": {
"input_per_1k": info.input_cost_per_1k,
"output_per_1k": info.output_cost_per_1k,
"estimated_per_request": self._estimate_typical_cost(info)
}
}
provider_info["models"].append(model_details)
result["summary"]["total_models"] += 1
# Track cheapest and largest
if result["summary"]["cheapest_model"] is None or \
info.input_cost_per_1k < result["summary"]["cheapest_model"]["cost"]:
result["summary"]["cheapest_model"] = {
"name": model_name,
"provider": provider,
"cost": info.input_cost_per_1k
}
if result["summary"]["largest_context"] is None or \
info.context_window > result["summary"]["largest_context"]["size"]:
result["summary"]["largest_context"] = {
"name": model_name,
"provider": provider,
"size": info.context_window
}
if provider_info["models"]:
result["providers"][provider] = provider_info
result["summary"]["providers_available"] += 1
return result
async def compare_models(self, model1: str, model2: str) -> Dict[str, Any]:
"""
Compare two models side by side
Args:
model1: First model name
model2: Second model name
Returns:
Detailed comparison
"""
comparison = self.manager.compare_models(model1, model2)
# Add recommendations
comparison["recommendations"] = self._generate_recommendations(comparison)
return comparison
async def suggest_model_for_task(self, task_type: str,
context_size: int,
budget: Optional[float] = None) -> Dict[str, Any]:
"""
Suggest best model for a specific task
Args:
task_type: Type of task (code_review, test_generation, etc.)
context_size: Estimated tokens needed
budget: Maximum cost in USD (optional)
Returns:
Model suggestion with reasoning
"""
# Task-specific requirements
task_requirements = {
'code_review': {
'min_context': 50000,
'needs_reasoning': True,
'preferred_providers': ['anthropic', 'openai']
},
'test_generation': {
'min_context': 20000,
'needs_code_gen': True,
'preferred_providers': ['anthropic', 'openai']
},
'documentation': {
'min_context': 10000,
'needs_formatting': True,
'preferred_providers': ['anthropic', 'openai', 'google']
},
'debugging': {
'min_context': 100000,
'needs_analysis': True,
'preferred_providers': ['anthropic', 'google']
},
'quick_task': {
'min_context': 5000,
'speed_priority': True,
'preferred_providers': ['openai', 'anthropic']
}
}
requirements = task_requirements.get(task_type, {
'min_context': context_size,
'preferred_providers': ['anthropic', 'openai', 'google']
})
# Find suitable models
candidates = []
for provider in requirements.get('preferred_providers', ['anthropic', 'openai']):
models = self.manager.FALLBACK_LIMITS.get(provider, {})
for model_name, limits in models.items():
if limits.get('context', 0) >= max(context_size, requirements.get('min_context', 0)):
info = await self.manager.get_model_info(model_name, provider)
# Calculate cost
cost = self.manager.estimate_cost(model_name, context_size, 2000)
# Check budget
if budget and cost > budget:
continue
candidates.append({
'model': model_name,
'provider': provider,
'context': info.context_window,
'cost': cost,
'score': self._score_model_for_task(info, requirements)
})
# Sort by score (best first)
candidates.sort(key=lambda x: x['score'], reverse=True)
if not candidates:
return {
'error': 'No suitable model found',
'suggestion': None,
'reason': f'No model can handle {context_size} tokens within budget'
}
best = candidates[0]
return {
'suggestion': best['model'],
'provider': best['provider'],
'context_window': best['context'],
'estimated_cost': best['cost'],
'reasoning': self._explain_choice(best, task_type, requirements),
'alternatives': candidates[1:3] if len(candidates) > 1 else []
}
async def get_current_model_status(self) -> Dict[str, Any]:
"""
Get current model being used and its status
Returns:
Current model information and usage statistics
"""
# Create a dynamic context manager to detect current model
dcm = DynamicContextManager(auto_detect=True)
current_model = dcm._detect_current_model()
# Get model info
info = await self.manager.get_model_info(current_model)
# Get cost estimate for typical usage
typical_cost = dcm.get_cost_estimate(10000, 2000)
return {
'current_model': current_model,
'provider': info.provider,
'capabilities': {
'context_window': info.context_window,
'max_output': info.max_output_tokens,
'usable_context': info.usable_context,
'supports_vision': info.supports_vision,
'supports_functions': info.supports_functions,
'supports_json_mode': info.supports_json_mode
},
'cost_estimate': typical_cost,
'context_usage': {
'tokens_used': dcm.current_tokens,
'tokens_available': dcm.get_remaining_tokens(),
'utilization': f"{(dcm.current_tokens / info.context_window * 100):.1f}%"
},
'recommendations': self._get_model_recommendations(info, dcm.current_tokens)
}
async def estimate_operation_cost(self, operation: str,
input_size: int,
output_size: int = 2000,
model: Optional[str] = None) -> Dict[str, Any]:
"""
Estimate cost for a specific operation
Args:
operation: Operation name (e.g., "code_review", "test_generation")
input_size: Input tokens
output_size: Expected output tokens
model: Specific model to use (optional)
Returns:
Cost breakdown and recommendations
"""
if not model:
# Detect current model
dcm = DynamicContextManager(auto_detect=True)
model = dcm._detect_current_model()
# Get cost for specified model
cost = self.manager.estimate_cost(model, input_size, output_size)
# Find cheaper alternatives
alternatives = []
for provider, models in self.manager.FALLBACK_LIMITS.items():
for model_name in models:
if model_name != model:
alt_cost = self.manager.estimate_cost(model_name, input_size, output_size)
if alt_cost < cost:
alternatives.append({
'model': model_name,
'provider': provider,
'cost': alt_cost,
'savings': cost - alt_cost,
'savings_percent': ((cost - alt_cost) / cost * 100) if cost > 0 else 0
})
# Sort by savings
alternatives.sort(key=lambda x: x['savings'], reverse=True)
return {
'operation': operation,
'model': model,
'tokens': {
'input': input_size,
'output': output_size,
'total': input_size + output_size
},
'cost': {
'total': cost,
'breakdown': {
'input': (input_size / 1000) * self.manager.PRICING.get(
model.split('-')[0] if '-' in model else model, {}
).get('input', 0.01),
'output': (output_size / 1000) * self.manager.PRICING.get(
model.split('-')[0] if '-' in model else model, {}
).get('output', 0.03)
}
},
'cheaper_alternatives': alternatives[:3],
'recommendation': self._get_cost_recommendation(cost, alternatives)
}
def _estimate_typical_cost(self, info) -> float:
"""Estimate cost for typical request (10k in, 2k out)"""
return (10 * info.input_cost_per_1k) + (2 * info.output_cost_per_1k)
def _score_model_for_task(self, info, requirements: Dict) -> float:
"""Score model based on task requirements"""
score = 100.0
# Context size fit
if 'min_context' in requirements:
if info.context_window >= requirements['min_context'] * 2:
score += 20 # Plenty of headroom
elif info.context_window >= requirements['min_context']:
score += 10 # Adequate
else:
score -= 50 # Insufficient
# Cost factor (lower is better)
score -= info.input_cost_per_1k * 10
# Feature requirements
if requirements.get('needs_vision') and not info.supports_vision:
score -= 100
if requirements.get('needs_json') and info.supports_json_mode:
score += 10
return max(0, score)
def _explain_choice(self, model_info: Dict, task_type: str,
requirements: Dict) -> str:
"""Explain why this model was chosen"""
reasons = []
if model_info['context'] >= requirements.get('min_context', 0) * 2:
reasons.append(f"Ample context window ({model_info['context']:,} tokens)")
if model_info['cost'] < 0.1:
reasons.append(f"Cost-effective (${model_info['cost']:.4f} per request)")
if 'anthropic' in model_info['provider'] and 'reasoning' in str(requirements):
reasons.append("Excellent reasoning capabilities")
if 'openai' in model_info['provider'] and 'code_gen' in str(requirements):
reasons.append("Strong code generation")
return f"Selected for {task_type}: " + ", ".join(reasons)
def _generate_recommendations(self, comparison: Dict) -> List[str]:
"""Generate recommendations based on model comparison"""
recs = []
comp = comparison.get('comparison', {})
if comp.get('context_ratio', 0) > 2:
recs.append(f"Use {comp['larger_context']} for large documents")
if 'cost_ratio' in comp and comp['cost_ratio'].get('input', 0) > 2:
recs.append(f"Use {comp['cheaper']} for cost-sensitive operations")
vision = comp.get('vision_support', {})
for model, has_vision in vision.items():
if has_vision:
recs.append(f"Use {model} for image/screenshot analysis")
break
return recs
def _get_model_recommendations(self, info, current_tokens: int) -> List[str]:
"""Get recommendations based on current usage"""
recs = []
utilization = current_tokens / info.context_window if info.context_window > 0 else 0
if utilization > 0.8:
recs.append("Consider using a model with larger context window")
elif utilization < 0.1 and info.input_cost_per_1k > 0.01:
recs.append("Consider using a cheaper model for small contexts")
if not info.supports_vision:
recs.append("Switch to vision-capable model for image analysis")
if info.input_cost_per_1k > 0.05:
recs.append("This is a premium model - ensure cost is justified")
return recs
def _get_cost_recommendation(self, cost: float, alternatives: List[Dict]) -> str:
"""Generate cost recommendation"""
if cost > 1.0:
return "⚠️ High cost operation - consider cheaper alternatives"
elif alternatives and alternatives[0]['savings_percent'] > 50:
return f"💰 You could save {alternatives[0]['savings_percent']:.0f}% with {alternatives[0]['model']}"
elif cost < 0.01:
return "✅ Very cost-effective operation"
else:
return "💵 Reasonable cost for this operation"