CHKIM79/scalable-ai-agent-system
0
1"""2Integration Layer3Provides RESTful APIs, webhooks, cloud connectors, and external service integrations4"""5import asyncio6import logging7import json8import aiohttp9import hmac10import hashlib11from typing import Dict, List, Any, Optional, Callable, Union12from dataclasses import dataclass, field13from enum import Enum14from datetime import datetime, timedelta15from urllib.parse import urljoin, urlparse16import sqlite317from fastapi import FastAPI, HTTPException, Request, BackgroundTasks18from fastapi.middleware.cors import CORSMiddleware19from pydantic import BaseModel, Field20import uvicorn21 22 23class IntegrationType(Enum):24 REST_API = "rest_api"25 WEBHOOK = "webhook"26 CLOUD_CONNECTOR = "cloud_connector"27 MESSAGE_QUEUE = "message_queue"28 DATABASE = "database"29 FILE_SYSTEM = "file_system"30 31 32class AuthType(Enum):33 NONE = "none"34 API_KEY = "api_key"35 BEARER_TOKEN = "bearer_token"36 OAUTH2 = "oauth2"37 BASIC_AUTH = "basic_auth"38 HMAC = "hmac"39 40 41@dataclass42class IntegrationConfig:43 id: str44 name: str45 integration_type: IntegrationType46 endpoint_url: str47 auth_type: AuthType = AuthType.NONE48 auth_config: Dict[str, Any] = field(default_factory=dict)49 headers: Dict[str, str] = field(default_factory=dict)50 timeout: int = 3051 retry_attempts: int = 352 rate_limit: Optional[int] = None53 enabled: bool = True54 metadata: Dict[str, Any] = field(default_factory=dict)55 56 57@dataclass58class WebhookConfig:59 id: str60 name: str61 endpoint: str62 secret: Optional[str] = None63 events: List[str] = field(default_factory=list)64 filters: Dict[str, Any] = field(default_factory=dict)65 retry_policy: Dict[str, Any] = field(default_factory=dict)66 enabled: bool = True67 68 69class APIRequest(BaseModel):70 endpoint: str71 method: str = "GET"72 headers: Optional[Dict[str, str]] = None73 params: Optional[Dict[str, Any]] = None74 data: Optional[Dict[str, Any]] = None75 timeout: Optional[int] = 3076 77 78class APIResponse(BaseModel):79 status_code: int80 headers: Dict[str, str]81 data: Any82 execution_time: float83 success: bool84 error: Optional[str] = None85 86 87class WebhookPayload(BaseModel):88 event: str89 data: Dict[str, Any]90 timestamp: datetime = Field(default_factory=datetime.now)91 source: str92 signature: Optional[str] = None93 94 95class CloudConnector:96 """Base class for cloud service connectors"""97 98 def __init__(self, config: IntegrationConfig):99 self.config = config100 self.session: Optional[aiohttp.ClientSession] = None101 self.logger = logging.getLogger(__name__)102 103 async def initialize(self):104 """Initialize connector"""105 self.session = aiohttp.ClientSession(106 timeout=aiohttp.ClientTimeout(total=self.config.timeout),107 headers=self.config.headers108 )109 110 async def authenticate(self) -> Dict[str, str]:111 """Perform authentication and return headers"""112 auth_headers = {}113 114 if self.config.auth_type == AuthType.API_KEY:115 key_name = self.config.auth_config.get('key_name', 'X-API-Key')116 api_key = self.config.auth_config.get('api_key')117 if api_key:118 auth_headers[key_name] = api_key119 120 elif self.config.auth_type == AuthType.BEARER_TOKEN:121 token = self.config.auth_config.get('token')122 if token:123 auth_headers['Authorization'] = f'Bearer {token}'124 125 elif self.config.auth_type == AuthType.BASIC_AUTH:126 username = self.config.auth_config.get('username')127 password = self.config.auth_config.get('password')128 if username and password:129 import base64130 credentials = base64.b64encode(f'{username}:{password}'.encode()).decode()131 auth_headers['Authorization'] = f'Basic {credentials}'132 133 return auth_headers134 135 async def make_request(self, 136 method: str, 137 endpoint: str, 138 data: Optional[Dict] = None,139 params: Optional[Dict] = None) -> APIResponse:140 """Make HTTP request to external service"""141 142 if not self.session:143 await self.initialize()144 145 # Prepare request146 url = urljoin(self.config.endpoint_url, endpoint)147 auth_headers = await self.authenticate()148 headers = {**self.config.headers, **auth_headers}149 150 start_time = datetime.now()151 152 try:153 async with self.session.request(154 method=method,155 url=url,156 json=data,157 params=params,158 headers=headers159 ) as response:160 161 execution_time = (datetime.now() - start_time).total_seconds()162 response_data = await response.json() if response.content_type == 'application/json' else await response.text()163 164 return APIResponse(165 status_code=response.status,166 headers=dict(response.headers),167 data=response_data,168 execution_time=execution_time,169 success=200 <= response.status < 300170 )171 172 except Exception as e:173 execution_time = (datetime.now() - start_time).total_seconds()174 self.logger.error(f"Request failed: {e}")175 176 return APIResponse(177 status_code=500,178 headers={},179 data=None,180 execution_time=execution_time,181 success=False,182 error=str(e)183 )184 185 async def shutdown(self):186 """Shutdown connector"""187 if self.session:188 await self.session.close()189 190 191class AWSConnector(CloudConnector):192 """AWS services connector"""193 194 async def authenticate(self) -> Dict[str, str]:195 """AWS authentication using access keys"""196 # Simplified AWS auth - in production use boto3197 access_key = self.config.auth_config.get('access_key_id')198 secret_key = self.config.auth_config.get('secret_access_key')199 200 if access_key and secret_key:201 # This is a simplified version - real AWS auth is more complex202 return {203 'Authorization': f'AWS {access_key}:{secret_key}',204 'X-Amz-Date': datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')205 }206 207 return await super().authenticate()208 209 async def invoke_lambda(self, function_name: str, payload: Dict) -> APIResponse:210 """Invoke AWS Lambda function"""211 return await self.make_request(212 method='POST',213 endpoint=f'/2015-03-31/functions/{function_name}/invocations',214 data=payload215 )216 217 async def put_s3_object(self, bucket: str, key: str, data: bytes) -> APIResponse:218 """Put object to S3"""219 # Simplified S3 put - real implementation would use proper AWS SDK220 return await self.make_request(221 method='PUT',222 endpoint=f'/{bucket}/{key}',223 data={'content': data.decode() if isinstance(data, bytes) else data}224 )225 226 227class GCPConnector(CloudConnector):228 """Google Cloud Platform connector"""229 230 async def authenticate(self) -> Dict[str, str]:231 """GCP authentication using service account"""232 service_account_key = self.config.auth_config.get('service_account_key')233 234 if service_account_key:235 # Simplified GCP auth - in production use google-auth236 return {237 'Authorization': f'Bearer {service_account_key}',238 'Content-Type': 'application/json'239 }240 241 return await super().authenticate()242 243 async def call_cloud_function(self, function_name: str, data: Dict) -> APIResponse:244 """Call Google Cloud Function"""245 return await self.make_request(246 method='POST',247 endpoint=f'/v1/projects/{self.config.auth_config.get("project_id")}/locations/{self.config.auth_config.get("region", "us-central1")}/functions/{function_name}:call',248 data=data249 )250 251 252class AzureConnector(CloudConnector):253 """Microsoft Azure connector"""254 255 async def authenticate(self) -> Dict[str, str]:256 """Azure authentication"""257 tenant_id = self.config.auth_config.get('tenant_id')258 client_id = self.config.auth_config.get('client_id')259 client_secret = self.config.auth_config.get('client_secret')260 261 if tenant_id and client_id and client_secret:262 # Simplified Azure auth - in production use azure-identity263 return {264 'Authorization': f'Bearer {client_secret}',265 'Content-Type': 'application/json'266 }267 268 return await super().authenticate()269 270 async def call_function_app(self, function_name: str, data: Dict) -> APIResponse:271 """Call Azure Function"""272 return await self.make_request(273 method='POST',274 endpoint=f'/api/{function_name}',275 data=data276 )277 278 279class WebhookManager:280 """Manages incoming and outgoing webhooks"""281 282 def __init__(self, db_path: str = "webhooks.db"):283 self.db_path = db_path284 self.webhooks: Dict[str, WebhookConfig] = {}285 self.event_handlers: Dict[str, List[Callable]] = {}286 self.db_connection = None287 self.logger = logging.getLogger(__name__)288 289 async def initialize(self):290 """Initialize webhook manager"""291 self.db_connection = sqlite3.connect(self.db_path)292 await self._create_tables()293 await self._load_webhooks()294 295 async def _create_tables(self):296 """Create webhook tables"""297 cursor = self.db_connection.cursor()298 299 cursor.execute('''300 CREATE TABLE IF NOT EXISTS webhooks (301 id TEXT PRIMARY KEY,302 name TEXT,303 endpoint TEXT,304 secret TEXT,305 events TEXT,306 filters TEXT,307 retry_policy TEXT,308 enabled BOOLEAN309 )310 ''')311 312 cursor.execute('''313 CREATE TABLE IF NOT EXISTS webhook_logs (314 id INTEGER PRIMARY KEY AUTOINCREMENT,315 webhook_id TEXT,316 event TEXT,317 payload TEXT,318 status TEXT,319 response TEXT,320 timestamp DATETIME,321 execution_time REAL322 )323 ''')324 325 self.db_connection.commit()326 327 async def _load_webhooks(self):328 """Load webhooks from database"""329 cursor = self.db_connection.cursor()330 cursor.execute('SELECT * FROM webhooks WHERE enabled = 1')331 332 for row in cursor.fetchall():333 webhook = WebhookConfig(334 id=row[0],335 name=row[1],336 endpoint=row[2],337 secret=row[3],338 events=json.loads(row[4]) if row[4] else [],339 filters=json.loads(row[5]) if row[5] else {},340 retry_policy=json.loads(row[6]) if row[6] else {},341 enabled=bool(row[7])342 )343 self.webhooks[webhook.id] = webhook344 345 def register_webhook(self, webhook: WebhookConfig):346 """Register a new webhook"""347 self.webhooks[webhook.id] = webhook348 349 # Persist to database350 cursor = self.db_connection.cursor()351 cursor.execute('''352 INSERT OR REPLACE INTO webhooks 353 (id, name, endpoint, secret, events, filters, retry_policy, enabled)354 VALUES (?, ?, ?, ?, ?, ?, ?, ?)355 ''', (356 webhook.id,357 webhook.name,358 webhook.endpoint,359 webhook.secret,360 json.dumps(webhook.events),361 json.dumps(webhook.filters),362 json.dumps(webhook.retry_policy),363 webhook.enabled364 ))365 self.db_connection.commit()366 367 self.logger.info(f"Registered webhook: {webhook.name}")368 369 def register_event_handler(self, event: str, handler: Callable):370 """Register event handler"""371 if event not in self.event_handlers:372 self.event_handlers[event] = []373 self.event_handlers[event].append(handler)374 375 async def process_incoming_webhook(self, request: Request) -> Dict[str, Any]:376 """Process incoming webhook request"""377 378 # Get request data379 body = await request.body()380 headers = dict(request.headers)381 382 # Verify signature if secret is provided383 webhook_id = request.path_params.get('webhook_id')384 if webhook_id and webhook_id in self.webhooks:385 webhook = self.webhooks[webhook_id]386 387 if webhook.secret:388 signature = headers.get('x-hub-signature-256', '')389 expected_signature = self._calculate_signature(body, webhook.secret)390 391 if not hmac.compare_digest(signature, expected_signature):392 raise HTTPException(status_code=401, detail="Invalid signature")393 394 # Parse payload395 try:396 payload_data = json.loads(body.decode())397 except json.JSONDecodeError:398 payload_data = {'raw_body': body.decode()}399 400 # Create webhook payload401 payload = WebhookPayload(402 event=payload_data.get('event', 'unknown'),403 data=payload_data,404 source=headers.get('user-agent', 'unknown'),405 signature=headers.get('x-hub-signature-256')406 )407 408 # Process event409 result = await self._process_event(payload)410 411 # Log webhook412 await self._log_webhook_event(webhook_id or 'unknown', payload, result)413 414 return result415 416 async def send_webhook(self, webhook_id: str, payload: WebhookPayload) -> APIResponse:417 """Send outgoing webhook"""418 419 if webhook_id not in self.webhooks:420 raise ValueError(f"Webhook {webhook_id} not found")421 422 webhook = self.webhooks[webhook_id]423 424 # Prepare request425 headers = {'Content-Type': 'application/json'}426 427 if webhook.secret:428 payload_json = json.dumps(payload.dict(), default=str)429 signature = self._calculate_signature(payload_json.encode(), webhook.secret)430 headers['X-Hub-Signature-256'] = signature431 432 # Send request433 start_time = datetime.now()434 435 try:436 async with aiohttp.ClientSession() as session:437 async with session.post(438 webhook.endpoint,439 json=payload.dict(),440 headers=headers,441 timeout=aiohttp.ClientTimeout(total=30)442 ) as response:443 444 execution_time = (datetime.now() - start_time).total_seconds()445 response_data = await response.text()446 447 result = APIResponse(448 status_code=response.status,449 headers=dict(response.headers),450 data=response_data,451 execution_time=execution_time,452 success=200 <= response.status < 300453 )454 455 # Log webhook456 await self._log_webhook_event(webhook_id, payload, result.dict())457 458 return result459 460 except Exception as e:461 execution_time = (datetime.now() - start_time).total_seconds()462 result = APIResponse(463 status_code=500,464 headers={},465 data=None,466 execution_time=execution_time,467 success=False,468 error=str(e)469 )470 471 await self._log_webhook_event(webhook_id, payload, result.dict())472 return result473 474 def _calculate_signature(self, payload: bytes, secret: str) -> str:475 """Calculate HMAC signature"""476 signature = hmac.new(477 secret.encode(),478 payload,479 hashlib.sha256480 ).hexdigest()481 return f'sha256={signature}'482 483 async def _process_event(self, payload: WebhookPayload) -> Dict[str, Any]:484 """Process webhook event"""485 486 results = []487 488 # Call registered handlers489 if payload.event in self.event_handlers:490 for handler in self.event_handlers[payload.event]:491 try:492 result = await handler(payload)493 results.append({494 'handler': handler.__name__,495 'success': True,496 'result': result497 })498 except Exception as e:499 results.append({500 'handler': handler.__name__,501 'success': False,502 'error': str(e)503 })504 505 return {506 'event': payload.event,507 'processed': len(results),508 'results': results509 }510 511 async def _log_webhook_event(self, webhook_id: str, payload: WebhookPayload, result: Dict):512 """Log webhook event"""513 cursor = self.db_connection.cursor()514 515 cursor.execute('''516 INSERT INTO webhook_logs 517 (webhook_id, event, payload, status, response, timestamp, execution_time)518 VALUES (?, ?, ?, ?, ?, ?, ?)519 ''', (520 webhook_id,521 payload.event,522 json.dumps(payload.dict(), default=str),523 'success' if result.get('success', False) else 'failed',524 json.dumps(result),525 datetime.now(),526 result.get('execution_time', 0.0)527 ))528 529 self.db_connection.commit()530 531 532class IntegrationLayer:533 """Main integration layer coordinating all external integrations"""534 535 def __init__(self, db_path: str = "integrations.db"):536 self.db_path = db_path537 self.integrations: Dict[str, IntegrationConfig] = {}538 self.connectors: Dict[str, CloudConnector] = {}539 self.webhook_manager = WebhookManager()540 541 self.db_connection = None542 self.logger = logging.getLogger(__name__)543 544 # Rate limiting545 self.rate_limits: Dict[str, List[datetime]] = {}546 547 async def initialize(self):548 """Initialize integration layer"""549 self.db_connection = sqlite3.connect(self.db_path)550 await self._create_tables()551 await self._load_integrations()552 await self.webhook_manager.initialize()553 554 self.logger.info("Integration layer initialized")555 556 async def _create_tables(self):557 """Create integration tables"""558 cursor = self.db_connection.cursor()559 560 cursor.execute('''561 CREATE TABLE IF NOT EXISTS integrations (562 id TEXT PRIMARY KEY,563 name TEXT,564 integration_type TEXT,565 endpoint_url TEXT,566 auth_type TEXT,567 auth_config TEXT,568 headers TEXT,569 timeout INTEGER,570 retry_attempts INTEGER,571 rate_limit INTEGER,572 enabled BOOLEAN,573 metadata TEXT574 )575 ''')576 577 cursor.execute('''578 CREATE TABLE IF NOT EXISTS integration_logs (579 id INTEGER PRIMARY KEY AUTOINCREMENT,580 integration_id TEXT,581 method TEXT,582 endpoint TEXT,583 status_code INTEGER,584 execution_time REAL,585 success BOOLEAN,586 error TEXT,587 timestamp DATETIME588 )589 ''')590 591 self.db_connection.commit()592 593 async def _load_integrations(self):594 """Load integrations from database"""595 cursor = self.db_connection.cursor()596 cursor.execute('SELECT * FROM integrations WHERE enabled = 1')597 598 for row in cursor.fetchall():599 config = IntegrationConfig(600 id=row[0],601 name=row[1],602 integration_type=IntegrationType(row[2]),603 endpoint_url=row[3],604 auth_type=AuthType(row[4]),605 auth_config=json.loads(row[5]) if row[5] else {},606 headers=json.loads(row[6]) if row[6] else {},607 timeout=row[7],608 retry_attempts=row[8],609 rate_limit=row[9],610 enabled=bool(row[10]),611 metadata=json.loads(row[11]) if row[11] else {}612 )613 614 await self.register_integration(config)615 616 async def register_integration(self, config: IntegrationConfig):617 """Register a new integration"""618 self.integrations[config.id] = config619 620 # Create appropriate connector621 if config.integration_type == IntegrationType.CLOUD_CONNECTOR:622 if 'aws' in config.endpoint_url.lower():623 connector = AWSConnector(config)624 elif 'googleapis' in config.endpoint_url.lower():625 connector = GCPConnector(config)626 elif 'azure' in config.endpoint_url.lower():627 connector = AzureConnector(config)628 else:629 connector = CloudConnector(config)630 631 await connector.initialize()632 self.connectors[config.id] = connector633 634 # Persist to database635 cursor = self.db_connection.cursor()636 cursor.execute('''637 INSERT OR REPLACE INTO integrations 638 (id, name, integration_type, endpoint_url, auth_type, auth_config, 639 headers, timeout, retry_attempts, rate_limit, enabled, metadata)640 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)641 ''', (642 config.id,643 config.name,644 config.integration_type.value,645 config.endpoint_url,646 config.auth_type.value,647 json.dumps(config.auth_config),648 json.dumps(config.headers),649 config.timeout,650 config.retry_attempts,651 config.rate_limit,652 config.enabled,653 json.dumps(config.metadata)654 ))655 self.db_connection.commit()656 657 self.logger.info(f"Registered integration: {config.name}")658 659 async def make_api_call(self, integration_id: str, request: APIRequest) -> APIResponse:660 """Make API call through integration"""661 662 if integration_id not in self.integrations:663 raise ValueError(f"Integration {integration_id} not found")664 665 config = self.integrations[integration_id]666 667 # Check rate limit668 if not await self._check_rate_limit(integration_id, config.rate_limit):669 return APIResponse(670 status_code=429,671 headers={},672 data=None,673 execution_time=0.0,674 success=False,675 error="Rate limit exceeded"676 )677 678 # Use connector if available679 if integration_id in self.connectors:680 connector = self.connectors[integration_id]681 response = await connector.make_request(682 method=request.method,683 endpoint=request.endpoint,684 data=request.data,685 params=request.params686 )687 else:688 # Direct HTTP call689 response = await self._make_direct_request(config, request)690 691 # Log the call692 await self._log_api_call(integration_id, request, response)693 694 return response695 696 async def _make_direct_request(self, config: IntegrationConfig, request: APIRequest) -> APIResponse:697 """Make direct HTTP request"""698 699 url = urljoin(config.endpoint_url, request.endpoint)700 headers = {**config.headers, **(request.headers or {})}701 702 # Add authentication703 if config.auth_type == AuthType.API_KEY:704 key_name = config.auth_config.get('key_name', 'X-API-Key')705 api_key = config.auth_config.get('api_key')706 if api_key:707 headers[key_name] = api_key708 709 start_time = datetime.now()710 711 try:712 async with aiohttp.ClientSession() as session:713 async with session.request(714 method=request.method,715 url=url,716 json=request.data,717 params=request.params,718 headers=headers,719 timeout=aiohttp.ClientTimeout(total=request.timeout or config.timeout)720 ) as response:721 722 execution_time = (datetime.now() - start_time).total_seconds()723 response_data = await response.json() if response.content_type == 'application/json' else await response.text()724 725 return APIResponse(726 status_code=response.status,727 headers=dict(response.headers),728 data=response_data,729 execution_time=execution_time,730 success=200 <= response.status < 300731 )732 733 except Exception as e:734 execution_time = (datetime.now() - start_time).total_seconds()735 return APIResponse(736 status_code=500,737 headers={},738 data=None,739 execution_time=execution_time,740 success=False,741 error=str(e)742 )743 744 async def _check_rate_limit(self, integration_id: str, rate_limit: Optional[int]) -> bool:745 """Check if request is within rate limit"""746 747 if not rate_limit:748 return True749 750 now = datetime.now()751 minute_ago = now - timedelta(minutes=1)752 753 # Clean old entries754 if integration_id in self.rate_limits:755 self.rate_limits[integration_id] = [756 timestamp for timestamp in self.rate_limits[integration_id]757 if timestamp > minute_ago758 ]759 else:760 self.rate_limits[integration_id] = []761 762 # Check limit763 if len(self.rate_limits[integration_id]) >= rate_limit:764 return False765 766 # Add current request767 self.rate_limits[integration_id].append(now)768 return True769 770 async def _log_api_call(self, integration_id: str, request: APIRequest, response: APIResponse):771 """Log API call"""772 cursor = self.db_connection.cursor()773 774 cursor.execute('''775 INSERT INTO integration_logs 776 (integration_id, method, endpoint, status_code, execution_time, success, error, timestamp)777 VALUES (?, ?, ?, ?, ?, ?, ?, ?)778 ''', (779 integration_id,780 request.method,781 request.endpoint,782 response.status_code,783 response.execution_time,784 response.success,785 response.error,786 datetime.now()787 ))788 789 self.db_connection.commit()790 791 def get_integration_stats(self, integration_id: str) -> Dict[str, Any]:792 """Get integration statistics"""793 cursor = self.db_connection.cursor()794 795 # Get recent calls796 cursor.execute('''797 SELECT COUNT(*), AVG(execution_time), 798 SUM(CASE WHEN success = 1 THEN 1 ELSE 0 END) as success_count799 FROM integration_logs 800 WHERE integration_id = ? AND timestamp > datetime('now', '-1 hour')801 ''', (integration_id,))802 803 row = cursor.fetchone()804 total_calls = row[0] or 0805 avg_time = row[1] or 0.0806 success_count = row[2] or 0807 808 success_rate = (success_count / total_calls) if total_calls > 0 else 0.0809 810 return {811 'integration_id': integration_id,812 'total_calls_last_hour': total_calls,813 'average_response_time': avg_time,814 'success_rate': success_rate,815 'current_rate_limit_usage': len(self.rate_limits.get(integration_id, []))816 }817 818 async def shutdown(self):819 """Shutdown integration layer"""820 # Shutdown connectors821 for connector in self.connectors.values():822 await connector.shutdown()823 824 if self.db_connection:825 self.db_connection.close()826 827 self.logger.info("Integration layer shutdown")828 829 830# FastAPI integration endpoints831def create_integration_api(integration_layer: IntegrationLayer) -> FastAPI:832 """Create FastAPI app for integration endpoints"""833 834 app = FastAPI(title="AI Agent Integration API", version="1.0.0")835 836 app.add_middleware(837 CORSMiddleware,838 allow_origins=["*"],839 allow_credentials=True,840 allow_methods=["*"],841 allow_headers=["*"],842 )843 844 @app.post("/api/call/{integration_id}")845 async def make_api_call(integration_id: str, request: APIRequest):846 """Make API call through integration"""847 try:848 response = await integration_layer.make_api_call(integration_id, request)849 return response.dict()850 except Exception as e:851 raise HTTPException(status_code=500, detail=str(e))852 853 @app.post("/webhook/{webhook_id}")854 async def receive_webhook(webhook_id: str, request: Request):855 """Receive incoming webhook"""856 try:857 result = await integration_layer.webhook_manager.process_incoming_webhook(request)858 return result859 except Exception as e:860 raise HTTPException(status_code=500, detail=str(e))861 862 @app.get("/api/integrations")863 async def list_integrations():864 """List all integrations"""865 return {866 integration_id: {867 'name': config.name,868 'type': config.integration_type.value,869 'enabled': config.enabled870 }871 for integration_id, config in integration_layer.integrations.items()872 }873 874 @app.get("/api/integrations/{integration_id}/stats")875 async def get_integration_stats(integration_id: str):876 """Get integration statistics"""877 try:878 stats = integration_layer.get_integration_stats(integration_id)879 return stats880 except Exception as e:881 raise HTTPException(status_code=500, detail=str(e))882 883 return app884