"""
MCP tools with Cequence analytics integration
"""
import time
import asyncio
from typing import Dict, Any
from datetime import datetime, timezone
from fastapi import APIRouter, Depends, Request
import structlog
from src.middleware.auth_middleware import get_auth_context, require_scopes
from src.core.descope_auth import AuthContext
from src.core.cequence_integration import track_agent_operation, track_security_event
logger = structlog.get_logger()
router = APIRouter()
@router.post("/tools/generate_application")
async def generate_application_tool(
request: Request,
project_description: str,
project_type: str = "fullstack",
auth: AuthContext = Depends(require_scopes("tools:generate"))
) -> Dict[str, Any]:
"""
Application generation tool with comprehensive monitoring
Required scope: tools:generate
"""
start_time = time.time()
correlation_id = getattr(request.state, 'correlation_id', 'unknown')
try:
logger.info(
"application_generation_started",
user_id=auth.user_id,
project_type=project_type,
description_length=len(project_description),
correlation_id=correlation_id
)
# Simulate application generation
await asyncio.sleep(2) # Placeholder for actual generation
# Calculate processing time
processing_time = (time.time() - start_time) * 1000
# Track successful operation
await track_agent_operation(
operation_type="application_generation",
agent_type="orchestrator",
correlation_id=correlation_id,
duration_ms=processing_time,
success=True,
metadata={
"project_type": project_type,
"description_length": len(project_description),
"user_id": auth.user_id
}
)
return {
"status": "success",
"project_id": correlation_id,
"project_type": project_type,
"estimated_completion": "5 minutes",
"processing_time_ms": processing_time,
"correlation_id": correlation_id
}
except Exception as e:
processing_time = (time.time() - start_time) * 1000
# Track failed operation
await track_agent_operation(
operation_type="application_generation",
agent_type="orchestrator",
correlation_id=correlation_id,
duration_ms=processing_time,
success=False,
metadata={
"error": str(e),
"user_id": auth.user_id
}
)
# Track security event for unexpected errors
await track_security_event(
event_type="generation_error",
severity="warning",
description=f"Application generation failed: {str(e)}",
correlation_id=correlation_id,
client_ip=request.client.host if request.client else "unknown",
user_id=auth.user_id
)
raise
@router.post("/tools/self_heal")
async def self_heal_tool(
request: Request,
project_id: str,
issue_description: str,
auth: AuthContext = Depends(require_scopes("tools:fix"))
) -> Dict[str, Any]:
"""
Self-healing tool with detailed operation tracking
Required scope: tools:fix
"""
start_time = time.time()
correlation_id = getattr(request.state, 'correlation_id', 'unknown')
try:
# Track healing attempt start
await track_agent_operation(
operation_type="self_healing_started",
agent_type="reviewer",
correlation_id=correlation_id,
duration_ms=0,
success=True,
metadata={
"project_id": project_id,
"issue_type": "automated_detection",
"user_id": auth.user_id
}
)
# Simulate healing process
await asyncio.sleep(3) # Placeholder for actual healing
processing_time = (time.time() - start_time) * 1000
# Track successful healing
await track_agent_operation(
operation_type="self_healing_completed",
agent_type="reviewer",
correlation_id=correlation_id,
duration_ms=processing_time,
success=True,
metadata={
"project_id": project_id,
"fixes_applied": 3,
"tests_passed": 5,
"user_id": auth.user_id
}
)
return {
"status": "healed",
"project_id": project_id,
"fixes_applied": 3,
"tests_passed": 5,
"processing_time_ms": processing_time,
"correlation_id": correlation_id
}
except Exception as e:
processing_time = (time.time() - start_time) * 1000
# Track failed healing
await track_agent_operation(
operation_type="self_healing_failed",
agent_type="reviewer",
correlation_id=correlation_id,
duration_ms=processing_time,
success=False,
metadata={
"project_id": project_id,
"error": str(e),
"user_id": auth.user_id
}
)
raise
@router.post("/tools/code_review")
async def code_review_tool(
request: Request,
project_id: str,
review_type: str = "security",
auth: AuthContext = Depends(require_scopes("tools:review"))
) -> Dict[str, Any]:
"""
Code review tool with analytics tracking
Required scope: tools:review
"""
start_time = time.time()
correlation_id = getattr(request.state, 'correlation_id', 'unknown')
try:
logger.info(
"code_review_started",
user_id=auth.user_id,
project_id=project_id,
review_type=review_type,
correlation_id=correlation_id
)
# Simulate code review
await asyncio.sleep(1.5)
processing_time = (time.time() - start_time) * 1000
# Track review operation
await track_agent_operation(
operation_type="code_review",
agent_type="reviewer",
correlation_id=correlation_id,
duration_ms=processing_time,
success=True,
metadata={
"project_id": project_id,
"review_type": review_type,
"issues_found": 2,
"severity": "medium",
"user_id": auth.user_id
}
)
return {
"status": "completed",
"project_id": project_id,
"review_type": review_type,
"issues_found": 2,
"severity": "medium",
"processing_time_ms": processing_time,
"correlation_id": correlation_id
}
except Exception as e:
processing_time = (time.time() - start_time) * 1000
await track_agent_operation(
operation_type="code_review",
agent_type="reviewer",
correlation_id=correlation_id,
duration_ms=processing_time,
success=False,
metadata={
"project_id": project_id,
"error": str(e),
"user_id": auth.user_id
}
)
raise
@router.post("/tools/test_generation")
async def test_generation_tool(
request: Request,
project_id: str,
test_type: str = "unit",
coverage_target: float = 90.0,
auth: AuthContext = Depends(require_scopes("tools:generate"))
) -> Dict[str, Any]:
"""
Test generation tool with performance tracking
Required scope: tools:generate
"""
start_time = time.time()
correlation_id = getattr(request.state, 'correlation_id', 'unknown')
try:
# Simulate test generation
await asyncio.sleep(2.5)
processing_time = (time.time() - start_time) * 1000
await track_agent_operation(
operation_type="test_generation",
agent_type="generator",
correlation_id=correlation_id,
duration_ms=processing_time,
success=True,
metadata={
"project_id": project_id,
"test_type": test_type,
"coverage_target": coverage_target,
"tests_generated": 15,
"coverage_achieved": 92.5,
"user_id": auth.user_id
}
)
return {
"status": "generated",
"project_id": project_id,
"test_type": test_type,
"tests_generated": 15,
"coverage_achieved": 92.5,
"processing_time_ms": processing_time,
"correlation_id": correlation_id
}
except Exception as e:
processing_time = (time.time() - start_time) * 1000
await track_agent_operation(
operation_type="test_generation",
agent_type="generator",
correlation_id=correlation_id,
duration_ms=processing_time,
success=False,
metadata={
"project_id": project_id,
"error": str(e),
"user_id": auth.user_id
}
)
raise
@router.post("/tools/deployment")
async def deployment_tool(
request: Request,
project_id: str,
environment: str = "staging",
auth: AuthContext = Depends(require_scopes("tools:deploy"))
) -> Dict[str, Any]:
"""
Deployment tool with environment tracking
Required scope: tools:deploy
"""
start_time = time.time()
correlation_id = getattr(request.state, 'correlation_id', 'unknown')
try:
# Simulate deployment
await asyncio.sleep(4)
processing_time = (time.time() - start_time) * 1000
await track_agent_operation(
operation_type="deployment",
agent_type="deployer",
correlation_id=correlation_id,
duration_ms=processing_time,
success=True,
metadata={
"project_id": project_id,
"environment": environment,
"deployment_url": f"https://{project_id}.{environment}.foundry.ai",
"health_check": "passed",
"user_id": auth.user_id
}
)
return {
"status": "deployed",
"project_id": project_id,
"environment": environment,
"deployment_url": f"https://{project_id}.{environment}.foundry.ai",
"health_check": "passed",
"processing_time_ms": processing_time,
"correlation_id": correlation_id
}
except Exception as e:
processing_time = (time.time() - start_time) * 1000
await track_agent_operation(
operation_type="deployment",
agent_type="deployer",
correlation_id=correlation_id,
duration_ms=processing_time,
success=False,
metadata={
"project_id": project_id,
"environment": environment,
"error": str(e),
"user_id": auth.user_id
}
)
raise
@router.get("/analytics/dashboard_data")
async def dashboard_data(
request: Request,
auth: AuthContext = Depends(require_scopes("admin:logs"))
) -> Dict[str, Any]:
"""
Provide dashboard data for demo presentation
Required scope: admin:logs
"""
correlation_id = getattr(request.state, 'correlation_id', 'unknown')
# Simulated dashboard metrics for demo - in production this would come from Cequence analytics
return {
"real_time_metrics": {
"active_requests": 12,
"requests_per_minute": 45,
"average_response_time_ms": 234,
"success_rate": 98.5,
"active_agents": {
"frontend": 2,
"backend": 3,
"reviewer": 1,
"deployer": 1
}
},
"security_status": {
"threat_level": "low",
"blocked_requests": 0,
"rate_limited_requests": 2,
"authentication_failures": 1,
"security_events_last_hour": 3
},
"generation_stats": {
"total_projects_generated": 157,
"successful_generations": 149,
"self_healing_events": 23,
"average_generation_time_seconds": 45,
"code_reviews_completed": 89,
"tests_generated": 1247
},
"top_users": [
{"user_id": "software_foundry_demo", "requests": 89},
{"user_id": "test_client_001", "requests": 23},
{"user_id": "integration_test", "requests": 12}
],
"recent_operations": [
{
"operation": "application_generation",
"user": "software_foundry_demo",
"status": "success",
"duration_ms": 2341,
"timestamp": datetime.now(timezone.utc).isoformat()
},
{
"operation": "self_healing",
"user": "test_client_001",
"status": "success",
"duration_ms": 3102,
"timestamp": datetime.now(timezone.utc).isoformat()
},
{
"operation": "code_review",
"user": "integration_test",
"status": "success",
"duration_ms": 1543,
"timestamp": datetime.now(timezone.utc).isoformat()
}
],
"performance_metrics": {
"gateway_latency_ms": 12,
"authentication_time_ms": 8,
"agent_orchestration_time_ms": 156,
"total_requests_today": 2847,
"errors_today": 23
},
"correlation_id": correlation_id,
"timestamp": datetime.now(timezone.utc).isoformat()
}
