async def connect_with_direct_token(
        self,
        db: AsyncSession,
        short_name: str,
        token: str,
        name: Optional[str],
        user: schemas.User,
        validate_token: bool = True,
    ) -> schemas.Connection:
        """Connect using a direct token (for local development).

        Args:
            db: The database session
            short_name: The short name of the integration
            token: The direct token
            name: The name of the connection
            user: The current user
            validate_token: Whether to validate the token

        Returns:
            The created connection

        Raises:
            HTTPException: If not in local development mode or validation fails
        """
        if not settings.LOCAL_DEVELOPMENT:
            raise HTTPException(
                status_code=403,
                detail="Direct token connection is only allowed in local development mode",
            )

        connection_name = name if name else f"Connection to {short_name.capitalize()}"

        # Service-specific token validation
        if validate_token:
            if short_name == "slack":
                connection_name = await self._validate_slack_token(token, name)

        async with UnitOfWork(db) as uow:
            # Get the integration
            source = await crud.source.get_by_short_name(uow.session, short_name)
            if not source:
                raise HTTPException(status_code=404, detail=f"{short_name} source not found")

            # Create integration credential for the token
            encrypted_creds = credentials.encrypt({"access_token": token})
            connection = await self._create_connection_with_credential(
                uow=uow,
                integration_type=IntegrationType.SOURCE,
                short_name=short_name,
                name=connection_name,
                integration_name=source.name,
                auth_type=AuthType.oauth2,  # We store it as OAuth2 for compatibility
                encrypted_credentials=encrypted_creds,
                auth_config_class=None,
                user=user,
            )

            await uow.commit()
            await uow.session.refresh(connection)
            return connection
Connect using a direct token (for local development).

        Args:
            db: The database session
            short_name: The short name of the integration
            token: The direct token
            name: The name of the connection
            user: The current user
            validate_token: Whether to validate the token

        Returns:
            The created connection

        Raises:
            HTTPException: If not in local development mode or validation fails
        
backend/airweave/core/connection_service.py