File size: 16,363 Bytes
5f505aa 035ac5e 82f9400 5f505aa |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 |
"""
Module: api.routes.agents
Description: Direct agent endpoints for specialized AI agents
Author: Anderson H. Silva
Date: 2025-09-25
License: Proprietary - All rights reserved
"""
from typing import Dict, Any, Optional
from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel, Field
from src.core import get_logger
from src.api.middleware.authentication import get_current_user
from src.api.auth import User
from src.infrastructure.rate_limiter import RateLimitTier
# Temporary function until proper rate limit tier detection is implemented
async def get_rate_limit_tier() -> RateLimitTier:
"""Get rate limit tier for current user."""
return RateLimitTier.BASIC
from src.agents import ZumbiAgent, AnitaAgent, TiradentesAgent, BonifacioAgent, AgentContext
from src.infrastructure.observability.metrics import track_time, count_calls
logger = get_logger(__name__)
router = APIRouter()
class AgentRequest(BaseModel):
"""Base request model for agent operations."""
query: str = Field(..., description="Query or input for the agent")
context: Dict[str, Any] = Field(default_factory=dict, description="Additional context")
options: Dict[str, Any] = Field(default_factory=dict, description="Agent-specific options")
class AgentResponse(BaseModel):
"""Base response model for agent operations."""
agent: str = Field(..., description="Agent name")
result: Dict[str, Any] = Field(..., description="Agent processing result")
metadata: Dict[str, Any] = Field(default_factory=dict, description="Processing metadata")
success: bool = Field(default=True, description="Operation success status")
message: Optional[str] = Field(None, description="Optional status message")
@router.post("/zumbi", response_model=AgentResponse)
@track_time("agent_zumbi_process")
@count_calls("agent_zumbi_requests")
async def process_zumbi_request(
request: AgentRequest,
background_tasks: BackgroundTasks,
current_user: User = Depends(get_current_user),
rate_limit_tier: RateLimitTier = Depends(get_rate_limit_tier)
):
"""
Process request with Zumbi dos Palmares agent.
Zumbi specializes in anomaly detection and investigation:
- Price anomalies and overpricing
- Vendor concentration and cartels
- Temporal pattern irregularities
- Contract duplication
- Payment irregularities
"""
try:
# Create agent context
context = AgentContext(
user_id=current_user.id if current_user else "anonymous",
session_id=str(request.context.get("session_id", "default")),
request_id=str(request.context.get("request_id", "unknown")),
metadata={
"rate_limit_tier": rate_limit_tier.value,
**request.context
}
)
# Initialize Zumbi agent
zumbi = ZumbiAgent()
# Process request
result = await zumbi.process(
message=request.query,
context=context,
**request.options
)
return AgentResponse(
agent="zumbi_dos_palmares",
result=result.data if hasattr(result, 'data') else {"analysis": str(result)},
metadata={
"processing_time": result.metadata.get("processing_time", 0) if hasattr(result, 'metadata') else 0,
"anomalies_detected": result.metadata.get("anomalies_detected", 0) if hasattr(result, 'metadata') else 0,
},
success=True,
message="Anomaly detection completed successfully"
)
except Exception as e:
logger.error(f"Zumbi agent error: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Zumbi agent processing failed: {str(e)}"
)
@router.post("/anita", response_model=AgentResponse)
@track_time("agent_anita_process")
@count_calls("agent_anita_requests")
async def process_anita_request(
request: AgentRequest,
background_tasks: BackgroundTasks,
current_user: User = Depends(get_current_user),
rate_limit_tier: RateLimitTier = Depends(get_rate_limit_tier)
):
"""
Process request with Anita Garibaldi agent.
Anita specializes in pattern analysis and correlation:
- Spending trends and patterns
- Organizational behavior analysis
- Vendor relationship mapping
- Seasonal pattern detection
- Efficiency metrics calculation
"""
try:
# Create agent context
context = AgentContext(
user_id=current_user.id if current_user else "anonymous",
session_id=str(request.context.get("session_id", "default")),
request_id=str(request.context.get("request_id", "unknown")),
metadata={
"rate_limit_tier": rate_limit_tier.value,
**request.context
}
)
# Initialize Anita agent
anita = AnitaAgent()
# Process request
result = await anita.process(
message=request.query,
context=context,
**request.options
)
return AgentResponse(
agent="anita_garibaldi",
result=result.data if hasattr(result, 'data') else {"analysis": str(result)},
metadata={
"processing_time": result.metadata.get("processing_time", 0) if hasattr(result, 'metadata') else 0,
"patterns_found": result.metadata.get("patterns_found", 0) if hasattr(result, 'metadata') else 0,
},
success=True,
message="Pattern analysis completed successfully"
)
except Exception as e:
logger.error(f"Anita agent error: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Anita agent processing failed: {str(e)}"
)
@router.post("/tiradentes", response_model=AgentResponse)
@track_time("agent_tiradentes_process")
@count_calls("agent_tiradentes_requests")
async def process_tiradentes_request(
request: AgentRequest,
background_tasks: BackgroundTasks,
current_user: User = Depends(get_current_user),
rate_limit_tier: RateLimitTier = Depends(get_rate_limit_tier)
):
"""
Process request with Tiradentes agent.
Tiradentes specializes in report generation:
- Executive summaries
- Detailed investigation reports
- Multi-format output (JSON, Markdown, HTML)
- Natural language explanations
- Actionable recommendations
"""
try:
# Create agent context
context = AgentContext(
user_id=current_user.id if current_user else "anonymous",
session_id=str(request.context.get("session_id", "default")),
request_id=str(request.context.get("request_id", "unknown")),
metadata={
"rate_limit_tier": rate_limit_tier.value,
**request.context
}
)
# Initialize Tiradentes agent
tiradentes = TiradentesAgent()
# Process request
result = await tiradentes.process(
message=request.query,
context=context,
**request.options
)
return AgentResponse(
agent="tiradentes",
result=result.data if hasattr(result, 'data') else {"report": str(result)},
metadata={
"processing_time": result.metadata.get("processing_time", 0) if hasattr(result, 'metadata') else 0,
"report_format": request.options.get("format", "markdown"),
},
success=True,
message="Report generation completed successfully"
)
except Exception as e:
logger.error(f"Tiradentes agent error: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Tiradentes agent processing failed: {str(e)}"
)
@router.post("/bonifacio", response_model=AgentResponse)
@track_time("agent_bonifacio_process")
@count_calls("agent_bonifacio_requests")
async def process_bonifacio_request(
request: AgentRequest,
background_tasks: BackgroundTasks,
current_user: User = Depends(get_current_user),
rate_limit_tier: RateLimitTier = Depends(get_rate_limit_tier)
):
"""
Process request with José Bonifácio agent.
Bonifácio specializes in legal and compliance analysis:
- Legal framework verification
- Compliance assessment
- Regulatory requirement checking
- Constitutional alignment analysis
- Legal risk identification
- Jurisprudence application
"""
try:
# Create agent context
context = AgentContext(
user_id=current_user.id if current_user else "anonymous",
session_id=str(request.context.get("session_id", "default")),
request_id=str(request.context.get("request_id", "unknown")),
metadata={
"rate_limit_tier": rate_limit_tier.value,
**request.context
}
)
# Initialize Bonifacio agent
bonifacio = BonifacioAgent()
# Process request
result = await bonifacio.process(
message=request.query,
context=context,
**request.options
)
return AgentResponse(
agent="jose_bonifacio",
result=result.data if hasattr(result, 'data') else {"analysis": str(result)},
metadata={
"processing_time": result.metadata.get("processing_time", 0) if hasattr(result, 'metadata') else 0,
"compliance_issues": result.metadata.get("compliance_issues", 0) if hasattr(result, 'metadata') else 0,
"legal_risks": result.metadata.get("legal_risks", []) if hasattr(result, 'metadata') else [],
},
success=True,
message="Legal and compliance analysis completed successfully"
)
except Exception as e:
logger.error(f"Bonifacio agent error: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Bonifacio agent processing failed: {str(e)}"
)
@router.post("/maria-quiteria", response_model=AgentResponse)
@track_time("agent_maria_quiteria_process")
@count_calls("agent_maria_quiteria_requests")
async def process_maria_quiteria_request(
request: AgentRequest,
background_tasks: BackgroundTasks,
current_user: User = Depends(get_current_user),
rate_limit_tier: RateLimitTier = Depends(get_rate_limit_tier)
):
"""
Process request with Maria Quitéria agent.
Maria Quitéria specializes in security auditing and system protection:
- Security threat detection
- Vulnerability assessment
- Compliance verification (LGPD, ISO27001, etc.)
- Intrusion detection
- Digital forensics
- Risk assessment
- Security monitoring
"""
try:
# Create agent context
context = AgentContext(
user_id=current_user.id if current_user else "anonymous",
session_id=str(uuid4()),
investigation_id=request.metadata.get("investigation_id", str(uuid4()))
)
# Get or create agent
from src.agents import MariaQuiteriaAgent
maria_quiteria = MariaQuiteriaAgent()
# Process request
result = await maria_quiteria.process(
message=request.query,
context=context,
**request.options
)
return AgentResponse(
agent="maria_quiteria",
result=result.data if hasattr(result, 'data') else {"analysis": str(result)},
metadata={
"processing_time": result.metadata.get("processing_time", 0) if hasattr(result, 'metadata') else 0,
"threats_detected": result.metadata.get("threats_detected", 0) if hasattr(result, 'metadata') else 0,
"security_score": result.metadata.get("security_score", 0) if hasattr(result, 'metadata') else 0,
},
success=True,
message="Security audit completed successfully"
)
except Exception as e:
logger.error(f"Maria Quiteria agent error: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Maria Quiteria agent processing failed: {str(e)}"
)
@router.get("/status")
async def get_agents_status(
current_user: User = Depends(get_current_user)
):
"""Get status of all available agents."""
agents = {
"zumbi_dos_palmares": {
"name": "Zumbi dos Palmares",
"role": "Anomaly Detection Specialist",
"status": "active",
"capabilities": [
"Price anomaly detection",
"Vendor concentration analysis",
"Temporal pattern recognition",
"Contract duplication detection",
"Payment irregularity identification"
]
},
"anita_garibaldi": {
"name": "Anita Garibaldi",
"role": "Pattern Analysis Specialist",
"status": "active",
"capabilities": [
"Spending trend analysis",
"Organizational behavior mapping",
"Vendor relationship analysis",
"Seasonal pattern detection",
"Efficiency metrics calculation"
]
},
"tiradentes": {
"name": "Tiradentes",
"role": "Report Generation Specialist",
"status": "active",
"capabilities": [
"Executive summary generation",
"Detailed investigation reports",
"Multi-format output",
"Natural language explanations",
"Actionable recommendations"
]
},
"jose_bonifacio": {
"name": "José Bonifácio",
"role": "Legal & Compliance Specialist",
"status": "active",
"capabilities": [
"Legal framework verification",
"Compliance assessment",
"Regulatory requirement checking",
"Constitutional alignment analysis",
"Legal risk identification"
]
},
"maria_quiteria": {
"name": "Maria Quitéria",
"role": "Security Auditor & System Guardian",
"status": "active",
"capabilities": [
"Security threat detection",
"Vulnerability assessment",
"Compliance verification (LGPD, ISO27001)",
"Intrusion detection",
"Digital forensics",
"Risk assessment"
]
}
}
return {
"agents": agents,
"total_active": sum(1 for a in agents.values() if a["status"] == "active"),
"timestamp": "2025-09-25T14:30:00Z"
}
@router.get("/")
async def list_agents():
"""List all available agents with their endpoints."""
return {
"message": "Cidadão.AI Agent System",
"version": "2.0.0",
"agents": [
{
"name": "Zumbi dos Palmares",
"endpoint": "/api/v1/agents/zumbi",
"description": "Anomaly detection and investigation specialist"
},
{
"name": "Anita Garibaldi",
"endpoint": "/api/v1/agents/anita",
"description": "Pattern analysis and correlation specialist"
},
{
"name": "Tiradentes",
"endpoint": "/api/v1/agents/tiradentes",
"description": "Report generation and natural language specialist"
},
{
"name": "José Bonifácio",
"endpoint": "/api/v1/agents/bonifacio",
"description": "Legal and compliance analysis specialist"
},
{
"name": "Maria Quitéria",
"endpoint": "/api/v1/agents/maria-quiteria",
"description": "Security auditing and system protection specialist"
}
]
} |