"""Connectors domain — HTTP routes. Router surfaces: - catalog_router GET /api/connectors global catalog (read-only) - workspace_defaults_router /api/workspaces/{id}/connector-defaults credential templates - project_router /api/projects/{id}/connectors project-scoped connections - capability_router /api/projects/{id}/capabilities PRIVATE Core invoke Public clients use Dashboard/BFF POST /api/projects/{id}/capabilities/{capability} """ from __future__ import annotations from uuid import UUID from fastapi import APIRouter, Depends, Header, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from src.core.database import get_db from src.core.dependencies import ( require_active_billing_for_project_id, require_active_billing_for_workspace_id, ) from src.core.exceptions import NotFoundError from src.domain.auth.models import User from src.domain.auth.router import get_current_user from src.domain.connectors.repository import ConnectorRepository, WorkspaceConnectorDefaultRepository from src.domain.connectors.schemas import ( ConnectorApprovalResponse, ConnectorConnectionResponse, ConnectorPermissionResponse, ConnectorResponse, CopyFromWorkspaceDefaultRequest, CreateConnectorApprovalRequest, CreateConnectorConnectionRequest, ExecuteConnectorActionRequest, ExecuteConnectorActionResponse, InvokeAutomationWorkflowTriggerRequest, InvokeAutomationWorkflowTriggerResponse, InvokeCrmCompanyRequest, InvokeCrmCompanyResponse, InvokeCrmDealRequest, InvokeCrmDealResponse, InvokeCrmAssociationRequest, InvokeCrmAssociationResponse, InvokeCrmContactRequest, InvokeCrmContactResponse, InvokeDocumentRequest, InvokeDocumentResponse, InvokeEmailSendRequest, InvokeEmailSendResponse, InvokeMessagingSmsSendRequest, InvokeMessagingSmsSendResponse, InvokePaymentsCaptureRequest, InvokePaymentsCatalogRequest, InvokePaymentsCheckoutRequest, InvokePaymentsRefundRequest, InvokePaymentsResponse, InvokeMessagingSendRequest, InvokeMessagingSendResponse, RotateConnectorSecretsRequest, UpdateConnectorConnectionRequest, UpsertConnectorPermissionRequest, UpsertWorkspaceConnectorDefaultRequest, WorkspaceConnectorDefaultResponse, ) from src.domain.cloud.dependencies import get_cloud_internal_service from src.domain.cloud.service import CloudInternalService from src.domain.connectors.service import ConnectorDefaultsService, ConnectorService from src.domain.projects.repository import ProjectRepository, ProjectSettingsRepository from src.integrations.cloud_control_client import CloudControlClient, get_cloud_client_for_workspace from src.integrations.platform_ports import enabled_capabilities as project_enabled_capabilities catalog_router = APIRouter(prefix="/api/connectors", tags=["connectors"]) workspace_defaults_router = APIRouter( prefix="/api/workspaces/{workspace_id}/connector-defaults", tags=["connectors"] ) project_router = APIRouter(prefix="/api/projects/{project_id}/connectors", tags=["connectors"]) capability_router = APIRouter(prefix="/api/projects/{project_id}/capabilities", tags=["capabilities"]) def _catalog_service(db: AsyncSession) -> ConnectorService: return ConnectorService( repo=ConnectorRepository(db), cloud_client=CloudControlClient(), ) async def _defaults_service(workspace_id: UUID, db: AsyncSession) -> ConnectorDefaultsService: cloud_client = await get_cloud_client_for_workspace(workspace_id, db) return ConnectorDefaultsService( defaults_repo=WorkspaceConnectorDefaultRepository(db), connector_repo=ConnectorRepository(db), cloud_client=cloud_client, ) async def _project_service( project_id: UUID, db: AsyncSession, ) -> tuple[ConnectorService, UUID]: project = await ProjectRepository(db).get_by_id(project_id) if not project: raise NotFoundError("Project not found") cloud_client = await get_cloud_client_for_workspace(project.workspace_id, db) return ( ConnectorService( repo=ConnectorRepository(db), cloud_client=cloud_client, ), project.workspace_id, ) # --------------------------------------------------------------------------- # Catalog (global, read-only) # --------------------------------------------------------------------------- @catalog_router.get("", response_model=list[ConnectorResponse]) async def list_connectors( _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> list[ConnectorResponse]: return await _catalog_service(db).list_catalog() @catalog_router.get("/{connector_key}", response_model=ConnectorResponse) async def get_connector( connector_key: str, _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorResponse: return await _catalog_service(db).get_catalog_connector(connector_key) # --------------------------------------------------------------------------- # Workspace connector defaults (credential templates — copy-on-demand) # --------------------------------------------------------------------------- @workspace_defaults_router.get("", response_model=list[WorkspaceConnectorDefaultResponse]) async def list_workspace_connector_defaults( workspace_id: UUID = Depends(require_active_billing_for_workspace_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> list[WorkspaceConnectorDefaultResponse]: svc = await _defaults_service(workspace_id, db) return await svc.list_defaults(workspace_id) @workspace_defaults_router.get( "/{connector_key}", response_model=WorkspaceConnectorDefaultResponse ) async def get_workspace_connector_default( connector_key: str, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> WorkspaceConnectorDefaultResponse: svc = await _defaults_service(workspace_id, db) return await svc.get_default(workspace_id, connector_key) @workspace_defaults_router.put( "/{connector_key}", response_model=WorkspaceConnectorDefaultResponse ) async def upsert_workspace_connector_default( connector_key: str, body: UpsertWorkspaceConnectorDefaultRequest, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> WorkspaceConnectorDefaultResponse: svc = await _defaults_service(workspace_id, db) return await svc.upsert_default(workspace_id, connector_key, body) @workspace_defaults_router.delete("/{connector_key}", status_code=204, response_model=None) async def delete_workspace_connector_default( connector_key: str, workspace_id: UUID = Depends(require_active_billing_for_workspace_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> None: svc = await _defaults_service(workspace_id, db) await svc.delete_default(workspace_id, connector_key) # --------------------------------------------------------------------------- # Project connections (all credentials/config/permissions scoped to project) # --------------------------------------------------------------------------- @project_router.get("", response_model=list[ConnectorConnectionResponse]) async def list_project_connector_connections( project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> list[ConnectorConnectionResponse]: svc, _workspace_id = await _project_service(project_id, db) return await svc.list_connections(project_id) @project_router.post("", response_model=ConnectorConnectionResponse, status_code=201) async def create_project_connector_connection( body: CreateConnectorConnectionRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorConnectionResponse: svc, workspace_id = await _project_service(project_id, db) return await svc.create_connection( workspace_id=workspace_id, project_id=project_id, body=body, ) @project_router.post( "/from-workspace-default", response_model=ConnectorConnectionResponse, status_code=201, ) async def copy_connector_from_workspace_default( body: CopyFromWorkspaceDefaultRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorConnectionResponse: """Copy a workspace credential template into a new project-scoped connection.""" svc, workspace_id = await _project_service(project_id, db) defaults_svc = ConnectorDefaultsService( defaults_repo=WorkspaceConnectorDefaultRepository(db), connector_repo=ConnectorRepository(db), cloud_client=svc._cloud, ) return await defaults_svc.copy_to_project( workspace_id=workspace_id, project_id=project_id, connector_key=body.connector_key, name=body.name, ) @project_router.get("/{connection_id}", response_model=ConnectorConnectionResponse) async def get_project_connector_connection( connection_id: UUID, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorConnectionResponse: svc, _workspace_id = await _project_service(project_id, db) return await svc.get_connection(project_id, connection_id) @project_router.patch("/{connection_id}", response_model=ConnectorConnectionResponse) async def update_project_connector_connection( connection_id: UUID, body: UpdateConnectorConnectionRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorConnectionResponse: svc, _workspace_id = await _project_service(project_id, db) return await svc.update_connection(project_id, connection_id, body) @project_router.delete("/{connection_id}", status_code=204, response_model=None) async def delete_project_connector_connection( connection_id: UUID, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> None: svc, _workspace_id = await _project_service(project_id, db) await svc.delete_connection(project_id, connection_id) @project_router.put("/{connection_id}/secrets", response_model=ConnectorConnectionResponse) async def rotate_project_connector_secrets( connection_id: UUID, body: RotateConnectorSecretsRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorConnectionResponse: svc, _workspace_id = await _project_service(project_id, db) return await svc.rotate_secrets(project_id, connection_id, body) @project_router.post( "/{connection_id}/permissions/allow-all", response_model=ConnectorConnectionResponse, ) async def allow_all_project_connector_permissions( connection_id: UUID, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorConnectionResponse: """Allow every action on a connection (fixes legacy deny-all rows).""" svc, _workspace_id = await _project_service(project_id, db) return await svc.allow_all_actions(project_id, connection_id) @project_router.put("/{connection_id}/permissions", response_model=ConnectorPermissionResponse) async def upsert_project_connector_permission( connection_id: UUID, body: UpsertConnectorPermissionRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorPermissionResponse: svc, _workspace_id = await _project_service(project_id, db) return await svc.upsert_permission(project_id, connection_id, body) @project_router.post("/{connection_id}/approvals", response_model=ConnectorApprovalResponse) async def create_project_connector_approval( connection_id: UUID, body: CreateConnectorApprovalRequest, project_id: UUID = Depends(require_active_billing_for_project_id), user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> ConnectorApprovalResponse: svc, workspace_id = await _project_service(project_id, db) return await svc.create_approval( workspace_id=workspace_id, project_id=project_id, connection_id=connection_id, user_id=user.id, body=body, ) @project_router.post("/{connection_id}/execute", response_model=ExecuteConnectorActionResponse, status_code=202) async def execute_project_connector_action( connection_id: UUID, body: ExecuteConnectorActionRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> ExecuteConnectorActionResponse: """Enqueue an integration.execute job for a connector action (dashboard/test use).""" from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) conn = await svc.get_connection(project_id, connection_id) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) result = await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, connection_id=connection_id, action=body.action, input=body.input, provider_hint=conn.connector_key, idempotency_key=body.idempotency_key, handshake_id="core_dashboard", ) if not result.get("ok"): code = result.get("reason_code") or "connector_action_denied" status = 502 if code == "provider_unavailable" else 403 raise HTTPException(status_code=status, detail=result.get("message") or code) job_id = str(result.get("job_id") or "") return ExecuteConnectorActionResponse( ok=True, job_id=job_id, status=str(result.get("status") or "queued"), poll_url=f"/api/projects/{project_id}/jobs/{job_id}" if job_id else None, ) _INVOKE_HTTP_STATUS = { "invalid_input": 400, "resource_unbound": 409, "text_required": 400, "destination_invalid": 400, "workflow_invalid": 400, "input_required": 400, "provider_fields_forbidden": 400, "destination_unbound": 409, "workflow_unbound": 409, "document_invalid": 400, "document_unbound": 409, "find_required": 400, "position_invalid": 400, "permission_denied": 403, "capability_not_bound": 409, "binding_ambiguous": 409, "connection_inactive": 409, "credentials_missing": 409, "capability_not_invocable": 409, "approval_required": 403, "invalid_approval_token": 403, "connector_action_denied": 403, "connection_not_found": 403, "scope_missing": 403, "capability_not_enabled": 403, "provider_unavailable": 502, "provider_rejected": 502, "email_provider_not_configured": 409, "sms_provider_not_configured": 409, "billing_required": 402, "crm_unbound": 409, "crm_contact_not_found": 404, "crm_company_not_found": 404, "crm_deal_not_found": 404, "crm_field_unsupported": 400, "crm_pipeline_unsupported": 400, "crm_pipeline_unbound": 409, "crm_stage_unbound": 409, "crm_association_unsupported": 400, "crm_association_invalid": 400, "crm_relationship_conflict": 409, "crm_duplicate_match": 409, "crm_validation_failed": 400, "crm_property_invalid": 400, "crm_permission_denied": 403, "reauthorization_required": 403, "crm_rate_limited": 429, "crm_provider_unavailable": 502, "cloud_credits_exhausted": 402, } @capability_router.post("/email.send", response_model=InvokeEmailSendResponse, status_code=202) async def invoke_project_email_send( body: InvokeEmailSendRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeEmailSendResponse: """Private email.send invoke: existing Cloud authorize + email.send enqueue.""" from src.domain.connectors.email_send_execute import enqueue_authorized_email_send from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") _svc, workspace_id = await _project_service(project_id, db) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) payload: dict = {"to": body.to, "subject": body.subject, "text": body.text} if body.html: payload["html"] = body.html if body.reply_to: payload["reply_to"] = body.reply_to result = await enqueue_authorized_email_send( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, payload=payload, idempotency_key=body.idempotency_key, ) if not result.get("ok"): code = result.get("reason_code") or "email_provider_not_configured" public = strip_capability_response( { "capability": "email.send", "status": "rejected", "reason_code": code, "message": result.get("message") or code, } ) raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(code, 409), detail=public, ) return InvokeEmailSendResponse( capability="email.send", status=str(result.get("status") or "queued"), message_id=str(result.get("job_id") or "") or None, ) @capability_router.post("/messaging.sms.send", response_model=InvokeMessagingSmsSendResponse, status_code=202) async def invoke_project_messaging_sms_send( body: InvokeMessagingSmsSendRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeMessagingSmsSendResponse: """Send SMS through user Twilio BYOK, or the provider-neutral P2HS default.""" from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.notification_sms import public_result_is_capability_shaped from src.domain.connectors.public_gateway import strip_capability_response from src.domain.connectors.sms_send_execute import enqueue_authorized_sms_send project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload = {"destination": body.destination, "text": body.text} result = await svc.invoke_notification_sms( workspace_id=workspace_id, project_id=project_id, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, approval_token=body.approval_token, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) if result.status == "queued": return InvokeMessagingSmsSendResponse(**result.as_dict()) # No eligible BYOK connection means the Core-controlled P2HS default gets a chance. # Explicit bindings/provider hints remain authoritative and must not silently fall back. if body.binding_connection_id is None and not body.provider_hint and result.reason_code == "capability_not_bound": fallback = await enqueue_authorized_sms_send( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, payload=payload, idempotency_key=body.idempotency_key, ) if fallback.get("ok"): return InvokeMessagingSmsSendResponse( capability="messaging.sms.send", status=str(fallback.get("status") or "queued"), message_id=str(fallback.get("job_id") or "") or None, ) result = type(result)( capability=result.capability, status="rejected", reason_code=str(fallback.get("reason_code") or "sms_provider_not_configured"), message=str(fallback.get("message") or "SMS delivery is not configured"), ) body_out = InvokeMessagingSmsSendResponse(**result.as_dict()) if body_out.status == "rejected": public = strip_capability_response(body_out.model_dump(exclude_none=True)) if not public_result_is_capability_shaped(public): public = {"capability": "messaging.sms.send", "status": "rejected", "reason_code": body_out.reason_code} raise HTTPException(status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=public) return body_out async def _invoke_payment_capability( *, capability: str, payload: dict, project_id: UUID, cloud_svc: CloudInternalService, db: AsyncSession, provider_hint: str | None, binding_connection_id: UUID | None, idempotency_key: str | None, ) -> InvokePaymentsResponse: from src.domain.connectors.payment_execute import enqueue_authorized_payment_job from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") _svc, workspace_id = await _project_service(project_id, db) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) result = await enqueue_authorized_payment_job( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, capability=capability, payload=payload, idempotency_key=idempotency_key, provider_hint=provider_hint, connection_id=str(binding_connection_id) if binding_connection_id else None, ) if not result.get("ok"): code = result.get("reason_code") or "capability_not_bound" public = strip_capability_response( { "capability": capability, "status": "rejected", "reason_code": code, "message": result.get("message") or code, } ) raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(code, 409), detail=public, ) return InvokePaymentsResponse( capability=capability, status=str(result.get("status") or "queued"), execution_ref=str(result.get("job_id") or "") or None, ) @capability_router.post( "/payment.checkout.create", response_model=InvokePaymentsResponse, status_code=202, ) @capability_router.post( "/payments.checkout.create", response_model=InvokePaymentsResponse, status_code=202, ) async def invoke_payments_checkout_create( body: InvokePaymentsCheckoutRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokePaymentsResponse: payload: dict = { "success_url": body.success_url, "cancel_url": body.cancel_url, "items": body.items, } if body.customer_email: payload["customer_email"] = body.customer_email if body.kind: payload["kind"] = body.kind if body.capture_mode: payload["capture_mode"] = body.capture_mode if body.related_order_id: payload["related_order_id"] = body.related_order_id if body.metadata is not None: payload["metadata"] = body.metadata return await _invoke_payment_capability( capability="payments.checkout.create", payload=payload, project_id=project_id, cloud_svc=cloud_svc, db=db, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, idempotency_key=body.idempotency_key, ) @capability_router.post( "/payment.capture", response_model=InvokePaymentsResponse, status_code=202, ) @capability_router.post( "/payments.order.capture", response_model=InvokePaymentsResponse, status_code=202, ) async def invoke_payments_order_capture( body: InvokePaymentsCaptureRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokePaymentsResponse: return await _invoke_payment_capability( capability="payments.order.capture", payload={"order_id": body.order_id}, project_id=project_id, cloud_svc=cloud_svc, db=db, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, idempotency_key=body.idempotency_key, ) @capability_router.post( "/payment.refund", response_model=InvokePaymentsResponse, status_code=202, ) @capability_router.post( "/payments.refund.create", response_model=InvokePaymentsResponse, status_code=202, ) async def invoke_payments_refund_create( body: InvokePaymentsRefundRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokePaymentsResponse: payload: dict = {"order_id": body.order_id} if body.amount is not None: payload["amount"] = body.amount return await _invoke_payment_capability( capability="payments.refund.create", payload=payload, project_id=project_id, cloud_svc=cloud_svc, db=db, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, idempotency_key=body.idempotency_key, ) @capability_router.post( "/payment.catalog.upsert", response_model=InvokePaymentsResponse, status_code=202, ) @capability_router.post( "/payments.catalog.upsert", response_model=InvokePaymentsResponse, status_code=202, ) async def invoke_payments_catalog_upsert( body: InvokePaymentsCatalogRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokePaymentsResponse: payload: dict = { "name": body.name, "unit_amount": body.unit_amount, "currency": body.currency, } if body.description: payload["description"] = body.description if body.sku: payload["sku"] = body.sku if body.product_record_id: payload["product_record_id"] = body.product_record_id if body.price_record_id: payload["price_record_id"] = body.price_record_id if body.interval: payload["interval"] = body.interval if body.active is not None: payload["active"] = body.active return await _invoke_payment_capability( capability="payments.catalog.upsert", payload=payload, project_id=project_id, cloud_svc=cloud_svc, db=db, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, idempotency_key=body.idempotency_key, ) @capability_router.post("/messaging.send", response_model=InvokeMessagingSendResponse, status_code=202) async def invoke_project_messaging_send( body: InvokeMessagingSendRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeMessagingSendResponse: """Private messaging.send invoke: resolve → authorize → integration.execute enqueue.""" from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload = {"destination": body.destination, "text": body.text} if body.thread_ref: payload["thread_ref"] = body.thread_ref if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_messaging_send( workspace_id=workspace_id, project_id=project_id, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, destination_map=body.destination_map, approval_token=body.approval_token, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) body_out = InvokeMessagingSendResponse(**result.as_dict()) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out @capability_router.post( "/automation.workflow.trigger", response_model=InvokeAutomationWorkflowTriggerResponse, status_code=202, ) async def invoke_project_automation_workflow_trigger( body: InvokeAutomationWorkflowTriggerRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeAutomationWorkflowTriggerResponse: """Private Make-backed invoke: resolve → authorize → integration.execute enqueue.""" from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload: dict = {"workflow": body.workflow, "input": body.input} if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_automation_workflow_trigger( workspace_id=workspace_id, project_id=project_id, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, workflow_map=body.workflow_map, approval_token=body.approval_token, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) public = strip_capability_response(result) body_out = InvokeAutomationWorkflowTriggerResponse(**public) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out async def _invoke_document_capability( *, capability: str, body: InvokeDocumentRequest, project_id: UUID, cloud_svc: CloudInternalService, db: AsyncSession, ) -> InvokeDocumentResponse: from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload: dict = {} if body.title is not None: payload["title"] = body.title if body.document is not None: payload["document"] = body.document if body.text is not None: payload["text"] = body.text if body.position is not None: payload["position"] = body.position if body.find is not None: payload["find"] = body.find if body.replace is not None: payload["replace"] = body.replace if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_document( workspace_id=workspace_id, project_id=project_id, capability=capability, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, document_map=body.document_map, approval_token=body.approval_token, enforce_invocable=True, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) public = strip_capability_response(result.as_dict()) body_out = InvokeDocumentResponse(**public) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out @capability_router.post( "/document.create", response_model=InvokeDocumentResponse, status_code=202, ) async def invoke_project_document_create( body: InvokeDocumentRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeDocumentResponse: """Private document.create: resolve → authorize → integration.execute.""" return await _invoke_document_capability( capability="document.create", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/document.get", response_model=InvokeDocumentResponse, status_code=202, ) async def invoke_project_document_get( body: InvokeDocumentRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeDocumentResponse: """Private document.get: resolve → authorize → integration.execute.""" return await _invoke_document_capability( capability="document.get", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/document.insert_text", response_model=InvokeDocumentResponse, status_code=202, ) async def invoke_project_document_insert_text( body: InvokeDocumentRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeDocumentResponse: """Private document.insert_text: resolve → authorize → integration.execute.""" return await _invoke_document_capability( capability="document.insert_text", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/document.append_text", response_model=InvokeDocumentResponse, status_code=202, ) async def invoke_project_document_append_text( body: InvokeDocumentRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeDocumentResponse: """Private document.append_text: resolve → authorize → integration.execute.""" return await _invoke_document_capability( capability="document.append_text", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/document.replace_text", response_model=InvokeDocumentResponse, status_code=202, ) async def invoke_project_document_replace_text( body: InvokeDocumentRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeDocumentResponse: """Private document.replace_text: resolve → authorize → integration.execute.""" return await _invoke_document_capability( capability="document.replace_text", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) async def _invoke_crm_contact_capability( *, capability: str, body: InvokeCrmContactRequest, project_id: UUID, cloud_svc: CloudInternalService, db: AsyncSession, ) -> InvokeCrmContactResponse: from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload: dict = {} if body.contact is not None: payload["contact"] = body.contact if body.changes is not None: payload["changes"] = body.changes if body.email is not None: payload["email"] = body.email if body.cursor is not None: payload["cursor"] = body.cursor if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_crm_contact( workspace_id=workspace_id, project_id=project_id, capability=capability, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, approval_token=body.approval_token, enforce_invocable=True, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) public = strip_capability_response(result.as_dict()) body_out = InvokeCrmContactResponse(**public) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out @capability_router.post( "/crm.contact.create", response_model=InvokeCrmContactResponse, status_code=202, ) async def invoke_project_crm_contact_create( body: InvokeCrmContactRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmContactResponse: """Private crm.contact.create: resolve → authorize → integration.execute.""" return await _invoke_crm_contact_capability( capability="crm.contact.create", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.contact.get", response_model=InvokeCrmContactResponse, status_code=202, ) async def invoke_project_crm_contact_get( body: InvokeCrmContactRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmContactResponse: return await _invoke_crm_contact_capability( capability="crm.contact.get", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.contact.update", response_model=InvokeCrmContactResponse, status_code=202, ) async def invoke_project_crm_contact_update( body: InvokeCrmContactRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmContactResponse: return await _invoke_crm_contact_capability( capability="crm.contact.update", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.contact.upsert", response_model=InvokeCrmContactResponse, status_code=202, ) async def invoke_project_crm_contact_upsert( body: InvokeCrmContactRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmContactResponse: return await _invoke_crm_contact_capability( capability="crm.contact.upsert", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.contact.search", response_model=InvokeCrmContactResponse, status_code=202, ) async def invoke_project_crm_contact_search( body: InvokeCrmContactRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmContactResponse: return await _invoke_crm_contact_capability( capability="crm.contact.search", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) async def _invoke_crm_company_capability( *, capability: str, body: InvokeCrmCompanyRequest, project_id: UUID, cloud_svc: CloudInternalService, db: AsyncSession, ) -> InvokeCrmCompanyResponse: from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload: dict = {} if body.company is not None: payload["company"] = body.company if body.changes is not None: payload["changes"] = body.changes if body.name is not None: payload["name"] = body.name if body.cursor is not None: payload["cursor"] = body.cursor if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_crm_company( workspace_id=workspace_id, project_id=project_id, capability=capability, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, approval_token=body.approval_token, enforce_invocable=True, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) public = strip_capability_response(result.as_dict()) body_out = InvokeCrmCompanyResponse(**public) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out @capability_router.post( "/crm.company.create", response_model=InvokeCrmCompanyResponse, status_code=202, ) async def invoke_project_crm_company_create( body: InvokeCrmCompanyRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmCompanyResponse: return await _invoke_crm_company_capability( capability="crm.company.create", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.company.update", response_model=InvokeCrmCompanyResponse, status_code=202, ) async def invoke_project_crm_company_update( body: InvokeCrmCompanyRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmCompanyResponse: return await _invoke_crm_company_capability( capability="crm.company.update", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.company.get", response_model=InvokeCrmCompanyResponse, status_code=202, ) async def invoke_project_crm_company_get( body: InvokeCrmCompanyRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmCompanyResponse: return await _invoke_crm_company_capability( capability="crm.company.get", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.company.search", response_model=InvokeCrmCompanyResponse, status_code=202, ) async def invoke_project_crm_company_search( body: InvokeCrmCompanyRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmCompanyResponse: return await _invoke_crm_company_capability( capability="crm.company.search", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) async def _invoke_crm_deal_capability( *, capability: str, body: InvokeCrmDealRequest, project_id: UUID, cloud_svc: CloudInternalService, db: AsyncSession, ) -> InvokeCrmDealResponse: from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload: dict = {} if body.deal is not None: payload["deal"] = body.deal if body.changes is not None: payload["changes"] = body.changes if body.name is not None: payload["name"] = body.name if body.stage is not None: payload["stage"] = body.stage if body.cursor is not None: payload["cursor"] = body.cursor if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_crm_deal( workspace_id=workspace_id, project_id=project_id, capability=capability, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, approval_token=body.approval_token, enforce_invocable=True, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) public = strip_capability_response(result.as_dict()) body_out = InvokeCrmDealResponse(**public) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out @capability_router.post( "/crm.deal.create", response_model=InvokeCrmDealResponse, status_code=202, ) async def invoke_project_crm_deal_create( body: InvokeCrmDealRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmDealResponse: return await _invoke_crm_deal_capability( capability="crm.deal.create", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.deal.update", response_model=InvokeCrmDealResponse, status_code=202, ) async def invoke_project_crm_deal_update( body: InvokeCrmDealRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmDealResponse: return await _invoke_crm_deal_capability( capability="crm.deal.update", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.deal.get", response_model=InvokeCrmDealResponse, status_code=202, ) async def invoke_project_crm_deal_get( body: InvokeCrmDealRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmDealResponse: return await _invoke_crm_deal_capability( capability="crm.deal.get", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) @capability_router.post( "/crm.deal.search", response_model=InvokeCrmDealResponse, status_code=202, ) async def invoke_project_crm_deal_search( body: InvokeCrmDealRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmDealResponse: return await _invoke_crm_deal_capability( capability="crm.deal.search", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) async def _invoke_crm_association_capability( *, capability: str, body: InvokeCrmAssociationRequest, project_id: UUID, cloud_svc: CloudInternalService, db: AsyncSession, ) -> InvokeCrmAssociationResponse: from src.domain.connectors.integration_execute import enqueue_authorized_integration_execute from src.domain.connectors.public_gateway import strip_capability_response project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) caps = project_enabled_capabilities(settings_row.settings or {}) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue_execute(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) payload: dict = {} if body.association is not None: payload["association"] = body.association if body.metadata is not None: payload["metadata"] = body.metadata result = await svc.invoke_crm_association( workspace_id=workspace_id, project_id=project_id, capability=capability, payload=payload, enabled_capabilities=caps, provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, approval_token=body.approval_token, enforce_invocable=True, enqueue_execute=enqueue_execute, idempotency_key=body.idempotency_key, ) public = strip_capability_response(result.as_dict()) body_out = InvokeCrmAssociationResponse(**public) if body_out.status == "rejected": raise HTTPException( status_code=_INVOKE_HTTP_STATUS.get(body_out.reason_code or "", 409), detail=body_out.model_dump(exclude_none=True), ) return body_out @capability_router.post( "/crm.association.create", response_model=InvokeCrmAssociationResponse, status_code=202, ) async def invoke_project_crm_association_create( body: InvokeCrmAssociationRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), ) -> InvokeCrmAssociationResponse: return await _invoke_crm_association_capability( capability="crm.association.create", body=body, project_id=project_id, cloud_svc=cloud_svc, db=db, ) def _register_resource_routes() -> None: from pydantic import create_model from src.domain.connectors.resource_contracts import RESOURCE_INPUTS def endpoint_for(verb, input_model): request_model = create_model( verb.replace(".", "_") + "Request", __base__=input_model, provider_hint=(str | None, None), binding_connection_id=(UUID | None, None), approval_token=(str | None, None), ) async def endpoint( body, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), cloud_svc: CloudInternalService = Depends(get_cloud_internal_service), db: AsyncSession = Depends(get_db), idempotency_key: str | None = Header(None, alias="Idempotency-Key"), ): from src.domain.connectors.integration_execute import ( enqueue_authorized_integration_execute, ) project = await ProjectRepository(db).get_by_id(project_id) if not project: raise HTTPException(status_code=404, detail="Project not found") svc, workspace_id = await _project_service(project_id, db) settings_row = await ProjectSettingsRepository(db).get_or_create(project_id) cloud_client = await get_cloud_client_for_workspace(workspace_id, db) async def enqueue(**kwargs): return await enqueue_authorized_integration_execute( cloud_svc=cloud_svc, cloud_client=cloud_client, project=project, workspace_id=workspace_id, handshake_id="core_capability_invoke", **kwargs, ) result = await svc.invoke_resource( workspace_id=workspace_id, project_id=project_id, capability=verb, payload=body.model_dump(exclude={"provider_hint", "binding_connection_id", "approval_token"}, exclude_none=True), provider_hint=body.provider_hint, binding_connection_id=body.binding_connection_id, approval_token=body.approval_token, idempotency_key=idempotency_key, enabled_capabilities=project_enabled_capabilities(settings_row.settings or {}), enqueue_execute=enqueue, ) if result["status"] == "rejected": reason = result.get("reason_code", "") raise HTTPException(status_code=_INVOKE_HTTP_STATUS.get(reason, 409), detail=result) return result endpoint.__annotations__["body"] = request_model endpoint.__name__ = "invoke_" + verb.replace(".", "_") return endpoint for verb, input_model in RESOURCE_INPUTS.items(): capability_router.add_api_route( f"/{verb}", endpoint_for(verb, input_model), methods=["POST"], status_code=202, ) _register_resource_routes()