feat: Instagram Graph API integration and custom OAuth2 implementation

backend/aci/common/schemas/security_scheme.py

@@ -79,6 +79,12 @@ class OAuth2Scheme(BaseModel):
     redirect_url: str | None = Field(
         default=None, min_length=1, max_length=2048, description="Redirect URL for OAuth2 callback."
     )
+    exchange_token_url: str | None = Field(
+        default=None,
+        min_length=1,
+        max_length=2048,
+        description="URL for exchanging short-lived tokens to long-lived tokens (Instagram specific)",
+    )
 
 
 # NOTE: need to show these fields for custom oauth2 app feature.

backend/aci/server/oauth2_manager.py

@@ -24,6 +24,7 @@ def __init__(
         access_token_url: str,
         refresh_token_url: str,
         token_endpoint_auth_method: str | None = None,
+        exchange_token_url: str | None = None,
     ):
         """
         Initialize the OAuth2Manager
@@ -39,6 +40,7 @@ def __init__(
             token_endpoint_auth_method:
                 client_secret_basic (default) | client_secret_post | none
                 Additional options can be achieved by registering a custom auth method
+            exchange_token_url: The URL for exchanging short-lived tokens to long-lived tokens (Instagram specific)
         """
         self.app_name = app_name
         self.client_id = client_id
@@ -48,6 +50,7 @@ def __init__(
         self.access_token_url = access_token_url
         self.refresh_token_url = refresh_token_url
         self.token_endpoint_auth_method = token_endpoint_auth_method
+        self.exchange_token_url = exchange_token_url
 
         # TODO: need to close the client after use
         # Add an aclose() helper (or implement __aenter__/__aexit__) and make callers invoke it during shutdown.
@@ -161,7 +164,48 @@ async def refresh_token(
             logger.error(f"Failed to refresh access token, app_name={self.app_name}, error={e}")
             raise OAuth2Error("Failed to refresh access token") from e
 
-    def parse_fetch_token_response(self, token: dict) -> OAuth2SchemeCredentials:
+    async def exchange_short_lived_token(self, short_lived_token: str) -> dict[str, Any]:
+        """
+        Exchange short-lived access token for long-lived access token.
+        This is specific to Instagram's API requirements.
+
+        Args:
+            short_lived_token: The short-lived access token from the initial OAuth flow
+
+        Returns:
+            Token response dictionary with long-lived access token
+        """
+        if self.app_name != "INSTAGRAM":
+            raise OAuth2Error("Token exchange is only supported for Instagram")
+
+        if not self.exchange_token_url:
+            raise OAuth2Error("Exchange token URL is not configured for Instagram")
+
+        try:
+            response = await self.oauth2_client.get(
+                self.exchange_token_url,
+                params={
+                    "grant_type": "ig_exchange_token",
+                    "client_secret": self.client_secret,
+                    "access_token": short_lived_token,
+                },
+                timeout=30.0,
+            )
+            response.raise_for_status()
+
+            token_data = cast(dict[str, Any], response.json())
+            logger.info(
+                f"Successfully exchanged short-lived token for long-lived token, app_name={self.app_name}"
+            )
+            return token_data
+
+        except Exception as e:
+            logger.error(
+                f"Failed to exchange short-lived token, app_name={self.app_name}, error={e}"
+            )
+            raise OAuth2Error("Failed to exchange short-lived token for long-lived token") from e
+
+    async def parse_fetch_token_response(self, token: dict) -> OAuth2SchemeCredentials:
         """
         Parse OAuth2SchemeCredentials from token response with app-specific handling.
 
@@ -181,6 +225,22 @@ def parse_fetch_token_response(self, token: dict) -> OAuth2SchemeCredentials:
                 logger.error(f"Missing authed_user in Slack OAuth response, app={self.app_name}")
                 raise OAuth2Error("Missing access_token in Slack OAuth response")
 
+        # handle Instagram's special case - exchange short-lived token for long-lived token
+        if self.app_name == "INSTAGRAM":
+            if "access_token" in data:
+                short_lived_token = data["access_token"]
+                logger.info(
+                    f"Exchanging short-lived token for long-lived token, app_name={self.app_name}"
+                )
+                long_lived_token_response = await self.exchange_short_lived_token(short_lived_token)
+                # Update data with long-lived token response: add expires_in and token_type, update access_token
+                data.update(long_lived_token_response)
+            else:
+                logger.error(
+                    f"Missing access_token in Instagram OAuth response, app={self.app_name}"
+                )
+                raise OAuth2Error("Missing access_token in Instagram OAuth response")
+
         if "access_token" not in data:
             logger.error(f"Missing access_token in OAuth response, app={self.app_name}")
             raise OAuth2Error("Missing access_token in OAuth response")

backend/aci/server/routes/linked_accounts.py

@@ -360,6 +360,7 @@ async def link_oauth2_account(
         access_token_url=oauth2_scheme.access_token_url,
         refresh_token_url=oauth2_scheme.refresh_token_url,
         token_endpoint_auth_method=oauth2_scheme.token_endpoint_auth_method,
+        exchange_token_url=oauth2_scheme.exchange_token_url,
     )
 
     path = request.url_for(LINKED_ACCOUNTS_OAUTH2_CALLBACK_ROUTE_NAME).path
@@ -433,6 +434,11 @@ async def linked_accounts_oauth2_callback(
 
     # check for state
     state_jwt = request.query_params.get("state")
+    # Special handling for Instagram: remove #_ suffix if present
+    if state_jwt and state_jwt.endswith("#_"):
+        state_jwt = state_jwt[:-2]  # Remove the last 2 characters (#_)
+        logger.info("Removed Instagram #_ suffix from state")
+
     if not state_jwt:
         logger.error(
             "OAuth2 account linking callback received, missing state",
@@ -501,14 +507,15 @@ async def linked_accounts_oauth2_callback(
         access_token_url=oauth2_scheme.access_token_url,
         refresh_token_url=oauth2_scheme.refresh_token_url,
         token_endpoint_auth_method=oauth2_scheme.token_endpoint_auth_method,
+        exchange_token_url=oauth2_scheme.exchange_token_url,
     )
 
     token_response = await oauth2_manager.fetch_token(
         redirect_uri=state.redirect_uri,
         code=code,
         code_verifier=state.code_verifier,
     )
-    security_credentials = oauth2_manager.parse_fetch_token_response(token_response)
+    security_credentials = await oauth2_manager.parse_fetch_token_response(token_response)
 
     # if the linked account already exists, update it, otherwise create a new one
     # TODO: consider separating the logic for updating and creating a linked account or give warning to clients

backend/aci/server/security_credentials_manager.py

@@ -17,6 +17,7 @@
     OAuth2SchemeCredentials,
     SecuritySchemeOverrides,
 )
+from aci.server import config
 from aci.server.oauth2_manager import OAuth2Manager
 
 logger = get_logger(__name__)
@@ -96,6 +97,17 @@ async def _get_oauth2_credentials(
         linked_account.security_credentials
     )
     if _access_token_is_expired(oauth2_scheme_credentials):
+        if app.name == "INSTAGRAM":
+            logger.error(
+                f"Access token expired, please re-authorize, linked_account_id={linked_account.id}, "
+                f"security_scheme={linked_account.security_scheme}, app={app.name}"
+            )
+            # NOTE: this error message could be used by the frontend to guide the user to re-authorize
+            raise OAuth2Error(
+                f"Access token expired. Please re-authorize at: "
+                f"{config.DEV_PORTAL_URL}/appconfigs/{app.name}"
+            )
+
         logger.warning(
             f"Access token expired, trying to refresh linked_account_id={linked_account.id}, "
             f"security_scheme={linked_account.security_scheme}, app={app.name}"
@@ -155,6 +167,7 @@ async def _refresh_oauth2_access_token(
         scope=oauth2_scheme_credentials.scope,
         authorize_url=oauth2_scheme.authorize_url,
         access_token_url=oauth2_scheme.access_token_url,
+        exchange_token_url=oauth2_scheme.exchange_token_url,
         refresh_token_url=oauth2_scheme.refresh_token_url,
         token_endpoint_auth_method=oauth2_scheme.token_endpoint_auth_method,
     )

backend/apps/instagram/app.json

@@ -0,0 +1,27 @@
+{
+    "name": "INSTAGRAM",
+    "display_name": "Instagram",
+    "logo": "https://raw.githubusercontent.com/aipotheosis-labs/aipolabs-icons/refs/heads/main/apps/instagram.svg",
+    "provider": "Meta Platforms, Inc.",
+    "version": "1.0.0",
+    "description": "The Instagram API allows developers to access and manage Instagram resources programmatically. It provides functionality for publishing content, retrieving user information, fetching post data, tracking user feeds, and managing direct messages through RESTful HTTP calls.",
+    "security_schemes": {
+      "oauth2": {
+        "location": "header",
+        "name": "Authorization",
+        "prefix": "Bearer",
+        "client_id": "{{ AIPOLABS_INSTAGRAM_APP_CLIENT_ID }}",
+        "client_secret": "{{ AIPOLABS_INSTAGRAM_APP_CLIENT_SECRET }}",
+        "scope": "instagram_business_basic instagram_business_content_publish instagram_business_manage_messages instagram_business_manage_comments instagram_business_manage_insights",
+        "authorize_url": "https://www.instagram.com/oauth/authorize",
+        "access_token_url": "https://api.instagram.com/oauth/access_token",
+        "exchange_token_url": "https://graph.instagram.com/access_token",
+        "refresh_token_url": "https://graph.instagram.com/refresh_access_token",
+        "token_endpoint_auth_method": "client_secret_post"
+      }
+    },
+    "default_security_credentials_by_scheme": {},
+    "categories": ["Social Media", "User Data"],
+    "visibility": "public",
+    "active": true
+  }