AuthProvider

from mcp.gumstack import AuthProvider, TokenResponse

class MyProvider(AuthProvider):
    name = "my-provider"  # Unique identifier
    
    def get_url(self, redirect_uri: str, state: str) -> str:
        """Return the OAuth authorization URL."""
        ...
    
    async def exchange(self, code: str, redirect_uri: str) -> TokenResponse:
        """Exchange authorization code for tokens."""
        ...
    
    async def refresh(self, refresh_token: str) -> TokenResponse:
        """Refresh an expired access token."""
        ...
    
    async def get_nickname(self, access_token: str) -> str:
        """Return display name for the connected account."""
        ...

async def revoke(self, access_token: str) -> bool:
    """Revoke the token. Return True if successful."""
    # Default: returns True (no-op)

from mcp.gumstack import TokenResponse

TokenResponse(
    access_token: str,           # Required
    refresh_token: str | None,   # For token refresh
    expires_in: int | None,      # Seconds until expiry
    token_type: str = "Bearer",
    scope: str | None = None
)

import os
import httpx
from mcp.gumstack import AuthProvider, TokenResponse

class LinearAuthProvider(AuthProvider):
    name = "linear"
    
    def get_url(self, redirect_uri: str, state: str) -> str:
        return (
            f"https://linear.app/oauth/authorize"
            f"?client_id={os.environ['LINEAR_CLIENT_ID']}"
            f"&redirect_uri={redirect_uri}"
            f"&state={state}"
            f"&response_type=code"
            f"&scope=read,write"
        )
    
    async def exchange(self, code: str, redirect_uri: str) -> TokenResponse:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.linear.app/oauth/token",
                data={
                    "grant_type": "authorization_code",
                    "client_id": os.environ["LINEAR_CLIENT_ID"],
                    "client_secret": os.environ["LINEAR_CLIENT_SECRET"],
                    "redirect_uri": redirect_uri,
                    "code": code,
                }
            )
            data = resp.json()
            return TokenResponse(
                access_token=data["access_token"],
                expires_in=data.get("expires_in")
            )
    
    async def refresh(self, refresh_token: str) -> TokenResponse:
        # Linear tokens don't expire, but implement if needed
        raise NotImplementedError("Linear tokens don't expire")
    
    async def get_nickname(self, access_token: str) -> str:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                "https://api.linear.app/graphql",
                headers={"Authorization": f"Bearer {access_token}"},
                json={"query": "{ viewer { email } }"}
            )
            return resp.json()["data"]["viewer"]["email"]