Skip to content

OAuth

octomate.oauth.base

What every provider's OAuth shares, below the flows themselves.

A credential is worth little at rest: it has to be carried to an upstream, and the upstream is the only thing that can say it has stopped working. That round trip belongs to no single provider, so it lives here rather than in one of them.

McpBearerAuth

Bases: Auth

Source code in octomate/oauth/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class McpBearerAuth(httpx2.Auth):
    def __init__(
        self,
        *,
        manager: OAuthManager,
        user: User,
        mcp_id: UUID7,
        connector_id: str,
        url: str,
    ) -> None:
        self.manager = manager
        self.user = user
        self.mcp_id = mcp_id
        self.connector_id = connector_id
        self.url = httpx2.URL(url)

    def headers(self, token: SecretStr) -> dict[str, str]:
        """Return the upstream's authentication headers for the user's token."""
        return {"Authorization": f"Bearer {token.get_secret_value()}"}

    async def async_auth_flow(
        self, request: httpx2.Request
    ) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
        if request.url.copy_with(query=None) != self.url:
            raise ValueError("OAuth credentials are bound to the installed MCP URL")
        token = await self.manager.access_token(
            self.user, self.connector_id, mcp_id=self.mcp_id
        )
        if token is None:
            raise ValueError(
                "This MCP needs authorization; connect it before calling its tools"
            )
        request.headers.update(self.headers(token))
        response = yield request
        if response.status_code == 401:
            await self.manager.invalidate(
                self.user, self.connector_id, mcp_id=self.mcp_id, expected_token=token
            )

manager instance-attribute 

manager = manager

user instance-attribute 

user = user

mcp_id instance-attribute 

mcp_id = mcp_id

connector_id instance-attribute 

connector_id = connector_id

url instance-attribute 

url = httpx2.URL(url)

headers 

headers(token: SecretStr) -> dict[str, str]

Return the upstream's authentication headers for the user's token.

Source code in octomate/oauth/base.py 
38
39
40
def headers(self, token: SecretStr) -> dict[str, str]:
    """Return the upstream's authentication headers for the user's token."""
    return {"Authorization": f"Bearer {token.get_secret_value()}"}

async_auth_flow async 

async_auth_flow(
    request: Request,
) -> AsyncGenerator[httpx2.Request, httpx2.Response]
Source code in octomate/oauth/base.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
async def async_auth_flow(
    self, request: httpx2.Request
) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
    if request.url.copy_with(query=None) != self.url:
        raise ValueError("OAuth credentials are bound to the installed MCP URL")
    token = await self.manager.access_token(
        self.user, self.connector_id, mcp_id=self.mcp_id
    )
    if token is None:
        raise ValueError(
            "This MCP needs authorization; connect it before calling its tools"
        )
    request.headers.update(self.headers(token))
    response = yield request
    if response.status_code == 401:
        await self.manager.invalidate(
            self.user, self.connector_id, mcp_id=self.mcp_id, expected_token=token
        )

octomate.oauth.flows

Standard OAuth flows shared by channels and MCP integrations.

logger module-attribute 

logger = logging.getLogger(__name__)

HTTPS_URL module-attribute 

HTTPS_URL: TypeAdapter[HttpsUrl] = TypeAdapter(HttpsUrl)

OAuthRefreshRejected

Bases: ValueError

The authorization server explicitly rejected the stored credentials.

Source code in octomate/oauth/flows.py 
41
42
class OAuthRefreshRejected(ValueError):
    """The authorization server explicitly rejected the stored credentials."""

DeviceCodeResponse pydantic-model

Bases: BaseModel

Fields:

Source code in octomate/oauth/flows.py 
45
46
47
48
49
50
51
class DeviceCodeResponse(BaseModel):
    device_code: SecretStr
    user_code: SecretStr
    verification_uri: HttpsUrl
    verification_uri_complete: HttpsUrl | None = None
    expires_in: int = Field(gt=0)
    interval: int = Field(default=5, ge=1)

device_code pydantic-field 

device_code: SecretStr

user_code pydantic-field 

user_code: SecretStr

verification_uri pydantic-field 

verification_uri: HttpsUrl

verification_uri_complete pydantic-field 

verification_uri_complete: HttpsUrl | None = None

expires_in pydantic-field 

expires_in: int

interval pydantic-field 

interval: int = 5

TokenError pydantic-model

Bases: BaseModel

Fields:

Source code in octomate/oauth/flows.py 
54
55
56
57
class TokenError(BaseModel):
    error: str
    error_description: str | None = None
    interval: int | None = Field(default=None, ge=1)

error pydantic-field 

error: str

error_description pydantic-field 

error_description: str | None = None

interval pydantic-field 

interval: int | None = None

OAuthTokenExchange

Source code in octomate/oauth/flows.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class OAuthTokenExchange:
    token_model: ClassVar[type[OAuthToken]] = OAuthToken
    response_adapter: TypeAdapter[OAuthToken | TokenError]
    authorization_endpoint: HttpsUrl | None
    token_endpoint: HttpsUrl
    client: OAuthClientInformationFull
    scopes: list[str]
    scope_separator: str
    invalid_credentials_errors: list[str]
    httpx_client_factory: McpHttpClientFactory
    discovery_state: OAuthDiscoveryState | None

    def __init__(
        self,
        *,
        authorization_endpoint: HttpsUrl | None = None,
        token_endpoint: HttpsUrl,
        client: OAuthClientInformationFull,
        scopes: list[str],
        scope_separator: str = " ",
        invalid_credentials_errors: list[str] | None = None,
        httpx_client_factory: McpHttpClientFactory,
        discovery_state: OAuthDiscoveryState | None = None,
    ) -> None:
        self.response_adapter = TypeAdapter(
            Annotated[
                Annotated[self.token_model, Tag("token")]
                | Annotated[TokenError, Tag("error")],
                Discriminator(token_response_kind),
            ]
        )
        self.authorization_endpoint = authorization_endpoint
        self.token_endpoint = token_endpoint
        self.client = client
        self.scopes = scopes
        self.scope_separator = scope_separator
        self.invalid_credentials_errors = (
            invalid_credentials_errors
            if invalid_credentials_errors is not None
            else ["invalid_grant", "invalid_client"]
        )
        self.httpx_client_factory = httpx_client_factory
        self.discovery_state = discovery_state

    @staticmethod
    async def send(
        client: httpx2.AsyncClient, request: httpx2.Request
    ) -> httpx2.Response:
        url = HTTPS_URL.validate_python(str(request.url))
        if url.username or url.password or url.fragment:
            raise ValueError(
                "OAuth endpoints cannot contain URL credentials or fragments"
            )
        response = await client.send(request, follow_redirects=False)
        if response.is_redirect:
            raise ValueError("OAuth endpoint redirects are not supported")
        return response

    async def request(self, url: HttpsUrl, data: dict[str, str]) -> httpx2.Response:
        if self.discovery_state is not None:
            check_registration_usable(self.client)
            if self.discovery_state.resource is not None:
                data["resource"] = str(self.discovery_state.resource)
        headers = {"Accept": "application/json"}
        match self.client.token_endpoint_auth_method:
            case "client_secret_basic":
                if self.client.client_secret is None:
                    raise ValueError("OAuth client secret is required")
                credentials = (
                    f"{quote_plus(self.client.client_id)}:"
                    f"{quote_plus(self.client.client_secret)}"
                )
                headers["Authorization"] = (
                    "Basic " + b64encode(credentials.encode()).decode()
                )
            case "client_secret_post":
                if self.client.client_secret is None:
                    raise ValueError("OAuth client secret is required")
                data["client_id"] = self.client.client_id
                data["client_secret"] = self.client.client_secret
            case "none" | None:
                data["client_id"] = self.client.client_id
            case _:
                raise ValueError(
                    "Unsupported OAuth token endpoint authentication method"
                )
        async with self.httpx_client_factory() as client:
            return await self.send(
                client,
                client.build_request("POST", str(url), data=data, headers=headers),
            )

    def parse_response(self, response: httpx2.Response) -> OAuthToken | TokenError:
        if response.status_code not in (200, 400, 401):
            response.raise_for_status()
        return self.response_adapter.validate_json(response.content)

    async def grant(
        self, response: httpx2.Response, token: OAuthToken | TokenError | None = None
    ) -> OAuthGrant:
        if token is None:
            token = self.parse_response(response)
        if isinstance(token, TokenError):
            logger.warning(
                "OAuth token request to %s failed (HTTP %s, %s): %s",
                self.token_endpoint.host,
                response.status_code,
                token.error,
                token.error_description,
            )
            if token.error in self.invalid_credentials_errors:
                raise OAuthRefreshRejected(
                    "OAuth credentials were rejected; reconnect this provider"
                )
            response.raise_for_status()
            raise ValueError(f"OAuth token request failed: {token.error}")
        response.raise_for_status()
        if token.expires_in is not None and token.expires_in < 0:
            raise ValueError("OAuth token lifetime cannot be negative")
        expires_at = calculate_token_expiry(token.expires_in)
        scopes = (
            [
                scope.strip()
                for scope in token.scope.split(self.scope_separator)
                if scope.strip()
            ]
            if token.scope is not None
            else self.scopes
        )
        return OAuthGrant(
            access_token=SecretStr(token.access_token),
            token_type=token.token_type,
            refresh_token=SecretStr(token.refresh_token)
            if token.refresh_token is not None
            else None,
            expires_at=datetime.fromtimestamp(expires_at, UTC)
            if expires_at is not None
            else None,
            scopes=scopes,
            discovery_state=self.discovery_state.model_copy(
                update={"scope": " ".join(scopes)}
            )
            if self.discovery_state is not None
            else None,
        )

    async def refresh(self, refresh_token: SecretStr) -> OAuthGrant:
        response = await self.request(
            self.token_endpoint,
            {
                "grant_type": "refresh_token",
                "refresh_token": refresh_token.get_secret_value(),
            },
        )
        grant = await self.grant(response)
        if grant.refresh_token is None:
            grant.refresh_token = refresh_token
        return grant

token_model class-attribute 

token_model: type[OAuthToken] = OAuthToken

response_adapter instance-attribute 

response_adapter: TypeAdapter[OAuthToken | TokenError] = (
    TypeAdapter(
        Annotated[
            Annotated[self.token_model, Tag("token")]
            | Annotated[TokenError, Tag("error")],
            Discriminator(token_response_kind),
        ]
    )
)

authorization_endpoint instance-attribute 

authorization_endpoint: HttpsUrl | None = (
    authorization_endpoint
)

token_endpoint instance-attribute 

token_endpoint: HttpsUrl = token_endpoint

client instance-attribute 

client: OAuthClientInformationFull = client

scopes instance-attribute 

scopes: list[str] = scopes

scope_separator instance-attribute 

scope_separator: str = scope_separator

invalid_credentials_errors instance-attribute 

invalid_credentials_errors: list[str] = (
    invalid_credentials_errors
    if invalid_credentials_errors is not None
    else ["invalid_grant", "invalid_client"]
)

httpx_client_factory instance-attribute 

httpx_client_factory: McpHttpClientFactory = (
    httpx_client_factory
)

discovery_state instance-attribute 

discovery_state: OAuthDiscoveryState | None = (
    discovery_state
)

send async staticmethod 

send(
    client: AsyncClient, request: Request
) -> httpx2.Response
Source code in octomate/oauth/flows.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
@staticmethod
async def send(
    client: httpx2.AsyncClient, request: httpx2.Request
) -> httpx2.Response:
    url = HTTPS_URL.validate_python(str(request.url))
    if url.username or url.password or url.fragment:
        raise ValueError(
            "OAuth endpoints cannot contain URL credentials or fragments"
        )
    response = await client.send(request, follow_redirects=False)
    if response.is_redirect:
        raise ValueError("OAuth endpoint redirects are not supported")
    return response

request async 

request(
    url: HttpsUrl, data: dict[str, str]
) -> httpx2.Response
Source code in octomate/oauth/flows.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def request(self, url: HttpsUrl, data: dict[str, str]) -> httpx2.Response:
    if self.discovery_state is not None:
        check_registration_usable(self.client)
        if self.discovery_state.resource is not None:
            data["resource"] = str(self.discovery_state.resource)
    headers = {"Accept": "application/json"}
    match self.client.token_endpoint_auth_method:
        case "client_secret_basic":
            if self.client.client_secret is None:
                raise ValueError("OAuth client secret is required")
            credentials = (
                f"{quote_plus(self.client.client_id)}:"
                f"{quote_plus(self.client.client_secret)}"
            )
            headers["Authorization"] = (
                "Basic " + b64encode(credentials.encode()).decode()
            )
        case "client_secret_post":
            if self.client.client_secret is None:
                raise ValueError("OAuth client secret is required")
            data["client_id"] = self.client.client_id
            data["client_secret"] = self.client.client_secret
        case "none" | None:
            data["client_id"] = self.client.client_id
        case _:
            raise ValueError(
                "Unsupported OAuth token endpoint authentication method"
            )
    async with self.httpx_client_factory() as client:
        return await self.send(
            client,
            client.build_request("POST", str(url), data=data, headers=headers),
        )

parse_response 

parse_response(
    response: Response,
) -> OAuthToken | TokenError
Source code in octomate/oauth/flows.py 
159
160
161
162
def parse_response(self, response: httpx2.Response) -> OAuthToken | TokenError:
    if response.status_code not in (200, 400, 401):
        response.raise_for_status()
    return self.response_adapter.validate_json(response.content)

grant async 

grant(
    response: Response,
    token: OAuthToken | TokenError | None = None,
) -> OAuthGrant
Source code in octomate/oauth/flows.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def grant(
    self, response: httpx2.Response, token: OAuthToken | TokenError | None = None
) -> OAuthGrant:
    if token is None:
        token = self.parse_response(response)
    if isinstance(token, TokenError):
        logger.warning(
            "OAuth token request to %s failed (HTTP %s, %s): %s",
            self.token_endpoint.host,
            response.status_code,
            token.error,
            token.error_description,
        )
        if token.error in self.invalid_credentials_errors:
            raise OAuthRefreshRejected(
                "OAuth credentials were rejected; reconnect this provider"
            )
        response.raise_for_status()
        raise ValueError(f"OAuth token request failed: {token.error}")
    response.raise_for_status()
    if token.expires_in is not None and token.expires_in < 0:
        raise ValueError("OAuth token lifetime cannot be negative")
    expires_at = calculate_token_expiry(token.expires_in)
    scopes = (
        [
            scope.strip()
            for scope in token.scope.split(self.scope_separator)
            if scope.strip()
        ]
        if token.scope is not None
        else self.scopes
    )
    return OAuthGrant(
        access_token=SecretStr(token.access_token),
        token_type=token.token_type,
        refresh_token=SecretStr(token.refresh_token)
        if token.refresh_token is not None
        else None,
        expires_at=datetime.fromtimestamp(expires_at, UTC)
        if expires_at is not None
        else None,
        scopes=scopes,
        discovery_state=self.discovery_state.model_copy(
            update={"scope": " ".join(scopes)}
        )
        if self.discovery_state is not None
        else None,
    )

refresh async 

refresh(refresh_token: SecretStr) -> OAuthGrant
Source code in octomate/oauth/flows.py 
213
214
215
216
217
218
219
220
221
222
223
224
async def refresh(self, refresh_token: SecretStr) -> OAuthGrant:
    response = await self.request(
        self.token_endpoint,
        {
            "grant_type": "refresh_token",
            "refresh_token": refresh_token.get_secret_value(),
        },
    )
    grant = await self.grant(response)
    if grant.refresh_token is None:
        grant.refresh_token = refresh_token
    return grant

OAuthTokenResolver

Bases: Protocol

Source code in octomate/oauth/flows.py 
227
228
class OAuthTokenResolver(Protocol):
    async def resolve(self, callback_uri: AnyHttpUrl | None) -> OAuthTokenExchange: ...

resolve async 

resolve(
    callback_uri: AnyHttpUrl | None,
) -> OAuthTokenExchange
Source code in octomate/oauth/flows.py 
228
async def resolve(self, callback_uri: AnyHttpUrl | None) -> OAuthTokenExchange: ...

AuthorizationCodeFlow

Authorization-code OAuth using a configured or discovered client.

Source code in octomate/oauth/flows.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
class AuthorizationCodeFlow:
    """Authorization-code OAuth using a configured or discovered client."""

    kind: ClassVar[Literal["authorization_code"]] = "authorization_code"
    tokens: OAuthTokenExchange | OAuthTokenResolver
    authorization_lifetime: timedelta

    def __init__(
        self,
        *,
        tokens: OAuthTokenExchange | OAuthTokenResolver,
        authorization_lifetime: timedelta,
    ) -> None:
        self.tokens = tokens
        self.authorization_lifetime = authorization_lifetime

    async def resolve_tokens(
        self, callback_uri: AnyHttpUrl | None = None
    ) -> OAuthTokenExchange:
        if isinstance(self.tokens, OAuthTokenExchange):
            return self.tokens
        return await self.tokens.resolve(callback_uri)

    async def start(
        self,
        context: OAuthFlowContext,
        callback_uri: AnyHttpUrl,
        state: SecretStr,
    ) -> AuthorizationRequest:
        tokens = await self.resolve_tokens(callback_uri)
        if tokens.authorization_endpoint is None:
            raise ValueError(
                "Authorization-code OAuth requires an authorization endpoint"
            )
        pkce = PKCEParameters.generate()
        params = {
            "response_type": "code",
            "client_id": tokens.client.client_id,
            "redirect_uri": str(callback_uri),
            "state": state.get_secret_value(),
            "code_challenge": pkce.code_challenge,
            "code_challenge_method": "S256",
        }
        if tokens.scopes:
            params["scope"] = " ".join(tokens.scopes)
        if tokens.discovery_state is not None:
            if tokens.discovery_state.resource is not None:
                params["resource"] = str(tokens.discovery_state.resource)
            if "offline_access" in tokens.scopes:
                params["prompt"] = "consent"
        return AuthorizationRequest(
            authorization_uri=AnyHttpUrl(
                str(
                    httpx2.URL(str(tokens.authorization_endpoint)).copy_merge_params(
                        params
                    )
                )
            ),
            code_verifier=SecretStr(pkce.code_verifier),
            expires_at=datetime.now(UTC) + self.authorization_lifetime,
            discovery_state=tokens.discovery_state,
        )

    async def exchange(
        self,
        context: OAuthFlowContext,
        *,
        code: str,
        code_verifier: SecretStr | None,
        callback_uri: AnyHttpUrl,
    ) -> OAuthGrant:
        if code_verifier is None:
            raise ValueError("OAuth requires a PKCE verifier")
        tokens = await self.resolve_tokens()
        response = await tokens.request(
            tokens.token_endpoint,
            {
                "grant_type": "authorization_code",
                "code": code,
                "code_verifier": code_verifier.get_secret_value(),
                "redirect_uri": str(callback_uri),
            },
        )
        return await tokens.grant(response)

    async def refresh(self, refresh_token: SecretStr) -> OAuthGrant:
        tokens = await self.resolve_tokens()
        return await tokens.refresh(refresh_token)

kind class-attribute 

kind: Literal['authorization_code'] = 'authorization_code'

tokens instance-attribute 

tokens: OAuthTokenExchange | OAuthTokenResolver = tokens

authorization_lifetime instance-attribute 

authorization_lifetime: timedelta = authorization_lifetime

resolve_tokens async 

resolve_tokens(
    callback_uri: AnyHttpUrl | None = None,
) -> OAuthTokenExchange
Source code in octomate/oauth/flows.py 
247
248
249
250
251
252
async def resolve_tokens(
    self, callback_uri: AnyHttpUrl | None = None
) -> OAuthTokenExchange:
    if isinstance(self.tokens, OAuthTokenExchange):
        return self.tokens
    return await self.tokens.resolve(callback_uri)

start async 

start(
    context: OAuthFlowContext,
    callback_uri: AnyHttpUrl,
    state: SecretStr,
) -> AuthorizationRequest
Source code in octomate/oauth/flows.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
async def start(
    self,
    context: OAuthFlowContext,
    callback_uri: AnyHttpUrl,
    state: SecretStr,
) -> AuthorizationRequest:
    tokens = await self.resolve_tokens(callback_uri)
    if tokens.authorization_endpoint is None:
        raise ValueError(
            "Authorization-code OAuth requires an authorization endpoint"
        )
    pkce = PKCEParameters.generate()
    params = {
        "response_type": "code",
        "client_id": tokens.client.client_id,
        "redirect_uri": str(callback_uri),
        "state": state.get_secret_value(),
        "code_challenge": pkce.code_challenge,
        "code_challenge_method": "S256",
    }
    if tokens.scopes:
        params["scope"] = " ".join(tokens.scopes)
    if tokens.discovery_state is not None:
        if tokens.discovery_state.resource is not None:
            params["resource"] = str(tokens.discovery_state.resource)
        if "offline_access" in tokens.scopes:
            params["prompt"] = "consent"
    return AuthorizationRequest(
        authorization_uri=AnyHttpUrl(
            str(
                httpx2.URL(str(tokens.authorization_endpoint)).copy_merge_params(
                    params
                )
            )
        ),
        code_verifier=SecretStr(pkce.code_verifier),
        expires_at=datetime.now(UTC) + self.authorization_lifetime,
        discovery_state=tokens.discovery_state,
    )

exchange async 

exchange(
    context: OAuthFlowContext,
    *,
    code: str,
    code_verifier: SecretStr | None,
    callback_uri: AnyHttpUrl,
) -> OAuthGrant
Source code in octomate/oauth/flows.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
async def exchange(
    self,
    context: OAuthFlowContext,
    *,
    code: str,
    code_verifier: SecretStr | None,
    callback_uri: AnyHttpUrl,
) -> OAuthGrant:
    if code_verifier is None:
        raise ValueError("OAuth requires a PKCE verifier")
    tokens = await self.resolve_tokens()
    response = await tokens.request(
        tokens.token_endpoint,
        {
            "grant_type": "authorization_code",
            "code": code,
            "code_verifier": code_verifier.get_secret_value(),
            "redirect_uri": str(callback_uri),
        },
    )
    return await tokens.grant(response)

refresh async 

refresh(refresh_token: SecretStr) -> OAuthGrant
Source code in octomate/oauth/flows.py 
316
317
318
async def refresh(self, refresh_token: SecretStr) -> OAuthGrant:
    tokens = await self.resolve_tokens()
    return await tokens.refresh(refresh_token)

DeviceAuthorizationFlow

OAuth device authorization with polling and token refresh.

Source code in octomate/oauth/flows.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
class DeviceAuthorizationFlow:
    """OAuth device authorization with polling and token refresh."""

    kind: ClassVar[Literal["device"]] = "device"
    device_authorization_endpoint: HttpsUrl
    tokens: OAuthTokenExchange

    def __init__(
        self,
        *,
        device_authorization_endpoint: HttpsUrl,
        tokens: OAuthTokenExchange,
    ) -> None:
        self.device_authorization_endpoint = device_authorization_endpoint
        self.tokens = tokens

    async def start(self, context: OAuthFlowContext) -> DeviceAuthorizationResponse:
        data = {"scope": " ".join(self.tokens.scopes)} if self.tokens.scopes else {}
        response = await self.tokens.request(self.device_authorization_endpoint, data)
        response.raise_for_status()
        authorization = DeviceCodeResponse.model_validate_json(response.content)
        return DeviceAuthorizationResponse(
            verification_uri=authorization.verification_uri,
            verification_uri_complete=authorization.verification_uri_complete,
            device_code=authorization.device_code,
            user_code=authorization.user_code,
            expires_at=datetime.now(UTC) + timedelta(seconds=authorization.expires_in),
            interval_seconds=authorization.interval,
        )

    async def complete(
        self, context: OAuthFlowContext, device_code: SecretStr
    ) -> OAuthGrant | OAuthPending:
        response = await self.tokens.request(
            self.tokens.token_endpoint,
            {
                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                "device_code": device_code.get_secret_value(),
            },
        )
        token = self.tokens.parse_response(response)
        if isinstance(token, TokenError):
            interval = max(context.interval_seconds or 5, token.interval or 5)
            match token.error:
                case "authorization_pending":
                    return OAuthPending(retry_after_seconds=interval)
                case "slow_down":
                    return OAuthPending(retry_after_seconds=interval + 5)
        return await self.tokens.grant(response, token)

    async def refresh(self, refresh_token: SecretStr) -> OAuthGrant:
        return await self.tokens.refresh(refresh_token)

kind class-attribute 

kind: Literal['device'] = 'device'

device_authorization_endpoint instance-attribute 

device_authorization_endpoint: HttpsUrl = (
    device_authorization_endpoint
)

tokens instance-attribute 

tokens: OAuthTokenExchange = tokens

start async 

start(
    context: OAuthFlowContext,
) -> DeviceAuthorizationResponse
Source code in octomate/oauth/flows.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
async def start(self, context: OAuthFlowContext) -> DeviceAuthorizationResponse:
    data = {"scope": " ".join(self.tokens.scopes)} if self.tokens.scopes else {}
    response = await self.tokens.request(self.device_authorization_endpoint, data)
    response.raise_for_status()
    authorization = DeviceCodeResponse.model_validate_json(response.content)
    return DeviceAuthorizationResponse(
        verification_uri=authorization.verification_uri,
        verification_uri_complete=authorization.verification_uri_complete,
        device_code=authorization.device_code,
        user_code=authorization.user_code,
        expires_at=datetime.now(UTC) + timedelta(seconds=authorization.expires_in),
        interval_seconds=authorization.interval,
    )

complete async 

complete(
    context: OAuthFlowContext, device_code: SecretStr
) -> OAuthGrant | OAuthPending
Source code in octomate/oauth/flows.py 
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
async def complete(
    self, context: OAuthFlowContext, device_code: SecretStr
) -> OAuthGrant | OAuthPending:
    response = await self.tokens.request(
        self.tokens.token_endpoint,
        {
            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
            "device_code": device_code.get_secret_value(),
        },
    )
    token = self.tokens.parse_response(response)
    if isinstance(token, TokenError):
        interval = max(context.interval_seconds or 5, token.interval or 5)
        match token.error:
            case "authorization_pending":
                return OAuthPending(retry_after_seconds=interval)
            case "slow_down":
                return OAuthPending(retry_after_seconds=interval + 5)
    return await self.tokens.grant(response, token)

refresh async 

refresh(refresh_token: SecretStr) -> OAuthGrant
Source code in octomate/oauth/flows.py 
371
372
async def refresh(self, refresh_token: SecretStr) -> OAuthGrant:
    return await self.tokens.refresh(refresh_token)

token_response_kind 

token_response_kind(
    value: OAuthToken | TokenError | JsonValue,
) -> str
Source code in octomate/oauth/flows.py 
60
61
62
63
64
def token_response_kind(value: OAuthToken | TokenError | JsonValue) -> str:
    # Some OAuth servers return protocol errors with HTTP 200.
    if isinstance(value, TokenError) or (isinstance(value, dict) and "error" in value):
        return "error"
    return "token"

octomate.oauth.mcp

MCP resource discovery and dynamic OAuth client registration.

McpOAuthDiscovery

Source code in octomate/oauth/mcp.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class McpOAuthDiscovery:
    url: HttpsUrl
    httpx_client_factory: McpHttpClientFactory
    client_metadata_url: HttpsUrl | None
    discovery_state: OAuthDiscoveryState | None

    def __init__(
        self,
        *,
        url: HttpsUrl,
        httpx_client_factory: McpHttpClientFactory,
        client_metadata_url: HttpsUrl | None = None,
        discovery_state: OAuthDiscoveryState | None = None,
    ) -> None:
        self.url = url
        self.httpx_client_factory = httpx_client_factory
        self.client_metadata_url = client_metadata_url
        self.discovery_state = discovery_state

    async def resolve(self, callback_uri: AnyHttpUrl | None) -> OAuthTokenExchange:
        state = (
            await self.discover(callback_uri)
            if callback_uri is not None
            else self.discovery_state
        )
        if state is None:
            raise ValueError("MCP authorization has no stored client registration")
        if (
            state.metadata.authorization_endpoint is None
            or state.metadata.token_endpoint is None
        ):
            raise ValueError("MCP authorization has no authorization-code endpoints")
        return OAuthTokenExchange(
            authorization_endpoint=HTTPS_URL.validate_python(
                str(state.metadata.authorization_endpoint)
            ),
            token_endpoint=HTTPS_URL.validate_python(
                str(state.metadata.token_endpoint)
            ),
            client=state.client,
            scopes=state.scope.split() if state.scope else [],
            httpx_client_factory=self.httpx_client_factory,
            discovery_state=state,
        )

    async def discover(self, callback_uri: AnyHttpUrl) -> OAuthDiscoveryState:
        server_url = str(self.url)
        # Pydantic URL equality includes the class; SDK registrations use AnyUrl.
        redirect_uri = AnyUrl(str(callback_uri))
        async with self.httpx_client_factory() as client:
            challenge = await OAuthTokenExchange.send(
                client,
                client.build_request(
                    "POST",
                    server_url,
                    headers={
                        "MCP-Protocol-Version": "2026-07-28",
                        "Accept": "application/json, text/event-stream",
                    },
                    json={
                        "jsonrpc": "2.0",
                        "id": "authorize",
                        "method": "tools/list",
                        "params": {},
                    },
                ),
            )
            if challenge.status_code != 401:
                challenge.raise_for_status()
                raise ValueError("MCP endpoint did not request OAuth authorization")
            resource = None
            for url in build_protected_resource_metadata_discovery_urls(
                extract_resource_metadata_from_www_auth(challenge), server_url
            ):
                response = await OAuthTokenExchange.send(
                    client, create_oauth_metadata_request(url)
                )
                if response.status_code == 404:
                    continue
                response.raise_for_status()
                resource = await handle_protected_resource_response(response)
                if resource is not None:
                    break
            if resource is None or not resource.authorization_servers:
                raise ValueError("MCP endpoint has no OAuth resource metadata")
            if not check_resource_allowed(
                requested_resource=resource_url_from_server_url(server_url),
                configured_resource=str(resource.resource),
            ):
                raise ValueError("OAuth resource does not match the installed MCP")
            issuer = str(resource.authorization_servers[0])
            metadata = None
            for url in build_oauth_authorization_server_metadata_discovery_urls(
                issuer, server_url
            ):
                response = await OAuthTokenExchange.send(
                    client, create_oauth_metadata_request(url)
                )
                if response.status_code == 404:
                    continue
                response.raise_for_status()
                ok, metadata = await handle_auth_metadata_response(response)
                if not ok:
                    break
                if metadata is not None:
                    validate_metadata_issuer(metadata, issuer)
                    break
            if (
                metadata is None
                or metadata.authorization_endpoint is None
                or metadata.token_endpoint is None
            ):
                raise ValueError("OAuth server has no authorization-code endpoints")
            HTTPS_URL.validate_python(str(metadata.authorization_endpoint))
            HTTPS_URL.validate_python(str(metadata.token_endpoint))
            if "S256" not in (metadata.code_challenge_methods_supported or []):
                raise ValueError("OAuth server must support PKCE S256")
            scope = get_client_metadata_scopes(
                extract_scope_from_www_auth(challenge),
                resource,
                metadata,
                ["authorization_code", "refresh_token"],
            )
            registration = OAuthClientMetadata(
                client_name="Octomate",
                redirect_uris=[redirect_uri],
                grant_types=["authorization_code", "refresh_token"],
                response_types=["code"],
                token_endpoint_auth_method="none",
                scope=scope,
            )
            if self.client_metadata_url is not None and should_use_client_metadata_url(
                metadata, str(self.client_metadata_url)
            ):
                info = create_client_info_from_metadata_url(
                    str(self.client_metadata_url), redirect_uris=[redirect_uri]
                )
            else:
                if metadata.registration_endpoint is None:
                    raise ValueError(
                        "This MCP requires pre-registered app configuration through a McpTentacle"
                    )
                response = await OAuthTokenExchange.send(
                    client,
                    create_client_registration_request(metadata, registration, issuer),
                )
                response.raise_for_status()
                info = await handle_registration_response(response)
            info.issuer = issuer
            check_registration_usable(info)
            if (
                info.client_secret_expires_at
                and info.client_secret_expires_at <= datetime.now(UTC).timestamp()
            ):
                raise ValueError("OAuth registration returned an expired client secret")
            info.validate_redirect_uri(redirect_uri)
        return OAuthDiscoveryState(
            resource=HTTPS_URL.validate_python(str(resource.resource)),
            metadata=metadata,
            client=info,
            scope=scope,
        )

url instance-attribute 

url: HttpsUrl = url

httpx_client_factory instance-attribute 

httpx_client_factory: McpHttpClientFactory = (
    httpx_client_factory
)

client_metadata_url instance-attribute 

client_metadata_url: HttpsUrl | None = client_metadata_url

discovery_state instance-attribute 

discovery_state: OAuthDiscoveryState | None = (
    discovery_state
)

resolve async 

resolve(
    callback_uri: AnyHttpUrl | None,
) -> OAuthTokenExchange
Source code in octomate/oauth/mcp.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def resolve(self, callback_uri: AnyHttpUrl | None) -> OAuthTokenExchange:
    state = (
        await self.discover(callback_uri)
        if callback_uri is not None
        else self.discovery_state
    )
    if state is None:
        raise ValueError("MCP authorization has no stored client registration")
    if (
        state.metadata.authorization_endpoint is None
        or state.metadata.token_endpoint is None
    ):
        raise ValueError("MCP authorization has no authorization-code endpoints")
    return OAuthTokenExchange(
        authorization_endpoint=HTTPS_URL.validate_python(
            str(state.metadata.authorization_endpoint)
        ),
        token_endpoint=HTTPS_URL.validate_python(
            str(state.metadata.token_endpoint)
        ),
        client=state.client,
        scopes=state.scope.split() if state.scope else [],
        httpx_client_factory=self.httpx_client_factory,
        discovery_state=state,
    )

discover async 

discover(callback_uri: AnyHttpUrl) -> OAuthDiscoveryState
Source code in octomate/oauth/mcp.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
async def discover(self, callback_uri: AnyHttpUrl) -> OAuthDiscoveryState:
    server_url = str(self.url)
    # Pydantic URL equality includes the class; SDK registrations use AnyUrl.
    redirect_uri = AnyUrl(str(callback_uri))
    async with self.httpx_client_factory() as client:
        challenge = await OAuthTokenExchange.send(
            client,
            client.build_request(
                "POST",
                server_url,
                headers={
                    "MCP-Protocol-Version": "2026-07-28",
                    "Accept": "application/json, text/event-stream",
                },
                json={
                    "jsonrpc": "2.0",
                    "id": "authorize",
                    "method": "tools/list",
                    "params": {},
                },
            ),
        )
        if challenge.status_code != 401:
            challenge.raise_for_status()
            raise ValueError("MCP endpoint did not request OAuth authorization")
        resource = None
        for url in build_protected_resource_metadata_discovery_urls(
            extract_resource_metadata_from_www_auth(challenge), server_url
        ):
            response = await OAuthTokenExchange.send(
                client, create_oauth_metadata_request(url)
            )
            if response.status_code == 404:
                continue
            response.raise_for_status()
            resource = await handle_protected_resource_response(response)
            if resource is not None:
                break
        if resource is None or not resource.authorization_servers:
            raise ValueError("MCP endpoint has no OAuth resource metadata")
        if not check_resource_allowed(
            requested_resource=resource_url_from_server_url(server_url),
            configured_resource=str(resource.resource),
        ):
            raise ValueError("OAuth resource does not match the installed MCP")
        issuer = str(resource.authorization_servers[0])
        metadata = None
        for url in build_oauth_authorization_server_metadata_discovery_urls(
            issuer, server_url
        ):
            response = await OAuthTokenExchange.send(
                client, create_oauth_metadata_request(url)
            )
            if response.status_code == 404:
                continue
            response.raise_for_status()
            ok, metadata = await handle_auth_metadata_response(response)
            if not ok:
                break
            if metadata is not None:
                validate_metadata_issuer(metadata, issuer)
                break
        if (
            metadata is None
            or metadata.authorization_endpoint is None
            or metadata.token_endpoint is None
        ):
            raise ValueError("OAuth server has no authorization-code endpoints")
        HTTPS_URL.validate_python(str(metadata.authorization_endpoint))
        HTTPS_URL.validate_python(str(metadata.token_endpoint))
        if "S256" not in (metadata.code_challenge_methods_supported or []):
            raise ValueError("OAuth server must support PKCE S256")
        scope = get_client_metadata_scopes(
            extract_scope_from_www_auth(challenge),
            resource,
            metadata,
            ["authorization_code", "refresh_token"],
        )
        registration = OAuthClientMetadata(
            client_name="Octomate",
            redirect_uris=[redirect_uri],
            grant_types=["authorization_code", "refresh_token"],
            response_types=["code"],
            token_endpoint_auth_method="none",
            scope=scope,
        )
        if self.client_metadata_url is not None and should_use_client_metadata_url(
            metadata, str(self.client_metadata_url)
        ):
            info = create_client_info_from_metadata_url(
                str(self.client_metadata_url), redirect_uris=[redirect_uri]
            )
        else:
            if metadata.registration_endpoint is None:
                raise ValueError(
                    "This MCP requires pre-registered app configuration through a McpTentacle"
                )
            response = await OAuthTokenExchange.send(
                client,
                create_client_registration_request(metadata, registration, issuer),
            )
            response.raise_for_status()
            info = await handle_registration_response(response)
        info.issuer = issuer
        check_registration_usable(info)
        if (
            info.client_secret_expires_at
            and info.client_secret_expires_at <= datetime.now(UTC).timestamp()
        ):
            raise ValueError("OAuth registration returned an expired client secret")
        info.validate_redirect_uri(redirect_uri)
    return OAuthDiscoveryState(
        resource=HTTPS_URL.validate_python(str(resource.resource)),
        metadata=metadata,
        client=info,
        scope=scope,
    )

octomate.oauth.routes

The browser half of an authorization-code connection.

Two routes, and deliberately only two: one that turns a UUID into the provider's real authorization request, and one the provider sends the user back to. Both are opened by a person's browser rather than called by anything — the authorization server never connects here, it only redirects the user agent — so what they answer with is a page for someone to read, and what they refuse with is the same page for every way of refusing.

Neither route authenticates its caller, because there is nobody to authenticate: the callback arrives out of band with no channel identity attached. Everything binding the round trip to one person was settled when the authorization started and is rechecked in OAuthManager.complete_callback.

logger module-attribute 

logger = logging.getLogger(__name__)

oauth_router module-attribute 

oauth_router = APIRouter(tags=['oauth'])

page 

page(
    title: str, detail: str, *, status_code: int = 200
) -> HTMLResponse

The whole user-facing surface of this router.

Says what happened and where the rest of the errand is, and nothing else — no provider name in an error, no operation id, no reason. Someone who reaches a refusal here is either the user, who needs to go back and ask again, or someone who should learn nothing from having tried.

Source code in octomate/oauth/routes.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def page(title: str, detail: str, *, status_code: int = 200) -> HTMLResponse:
    """The whole user-facing surface of this router.

    Says what happened and where the rest of the errand is, and nothing else — no
    provider name in an error, no operation id, no reason. Someone who reaches a
    refusal here is either the user, who needs to go back and ask again, or someone
    who should learn nothing from having tried.
    """
    return HTMLResponse(
        "<!doctype html>"
        '<html lang="en"><head><meta charset="utf-8">'
        '<meta name="viewport" content="width=device-width,initial-scale=1">'
        f"<title>{title}</title>"
        "<style>body{font:16px/1.6 system-ui,sans-serif;margin:0;display:grid;"
        "place-items:center;min-height:100vh;padding:2rem;color:#1c1c1e;"
        "background:#fbfbfd}main{max-width:26rem;text-align:center}"
        "h1{font-size:1.25rem;margin:0 0 .5rem}p{margin:0;color:#5f5f66}"
        "@media(prefers-color-scheme:dark){body{color:#f2f2f7;background:#131315}"
        "p{color:#a1a1aa}}</style></head>"
        f"<body><main><h1>{title}</h1><p>{detail}</p></main></body></html>",
        status_code=status_code,
    )

start async 

start(
    connector_id: str,
    operation_id: UUID7,
    manager: Annotated[
        OAuthManager, Depends(oauth_manager)
    ],
) -> Response
Source code in octomate/oauth/routes.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@oauth_router.get(OAUTH_START_PATH, include_in_schema=False)
async def start(
    connector_id: str,
    operation_id: UUID7,
    manager: Annotated[OAuthManager, Depends(oauth_manager)],
) -> Response:
    try:
        staged = await manager.staged_authorization(connector_id, operation_id)
    except UnusableOAuthOperation as error:
        logger.info("Refused an OAuth start link: %s", error)
        return page(
            "This link has expired",
            "Ask again in the chat where you requested it and a fresh one will arrive.",
            status_code=404,
        )
    # 307 rather than 302: a browser that turned this into a GET would be harmless
    # today, but the provider request is this operation's only chance to be made
    # exactly as it was staged.
    return RedirectResponse(str(staged.authorization_uri), status_code=307)

callback async 

callback(
    connector_id: str,
    manager: Annotated[
        OAuthManager, Depends(oauth_manager)
    ],
    state: Annotated[str, Query()] = "",
    code: Annotated[str, Query()] = "",
    error: Annotated[str, Query()] = "",
    iss: Annotated[str | None, Query()] = None,
) -> HTMLResponse | RedirectResponse
Source code in octomate/oauth/routes.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@oauth_router.get(OAUTH_CALLBACK_PATH, include_in_schema=False, response_model=None)
async def callback(
    connector_id: str,
    manager: Annotated[OAuthManager, Depends(oauth_manager)],
    state: Annotated[str, Query()] = "",
    code: Annotated[str, Query()] = "",
    error: Annotated[str, Query()] = "",
    iss: Annotated[str | None, Query()] = None,
) -> HTMLResponse | RedirectResponse:
    if not state:
        return page(
            "Something went wrong",
            "This page was opened without an authorization to finish.",
            status_code=400,
        )
    if error or not code:
        # A provider that says no still says which authorization it is saying no
        # about, so the operation is closed rather than left live until it ages out.
        # Failing to close it is not worth telling the user about.
        try:
            await manager.abandon_callback(connector_id, state=state, issuer=iss)
        except (ValueError, OAuthFlowError):
            return page(
                "This link is invalid",
                "The authorization response could not be verified.",
                status_code=400,
            )
        return page(
            "Not connected",
            "The authorization was declined. You can ask again in the chat any time.",
        )
    try:
        completed = await manager.complete_callback(
            connector_id, state=state, code=code, issuer=iss
        )
    except UnusableOAuthOperation as unusable:
        logger.info("Refused an OAuth callback: %s", unusable)
        return page(
            "This link has expired",
            "Ask again in the chat where you requested it and a fresh one will arrive.",
            status_code=404,
        )
    except OAuthFlowError:
        return page(
            "This link is invalid",
            "The authorization response could not be verified.",
            status_code=400,
        )
    except Exception as failure:
        # The provider refused the exchange, or never answered. The operation is
        # spent either way, so there is nothing to do but say so and let them ask
        # for another.
        logger.warning(
            "Failed to complete an OAuth callback (%s)", type(failure).__name__
        )
        return page(
            "Something went wrong",
            "The connection could not be completed. Ask again in the chat and a "
            "fresh link will arrive.",
            status_code=502,
        )
    grant = completed.grant
    title = (
        f"Connected as {escape(grant.account_label)}"
        if grant.account_label
        else "Connected"
    )
    try:
        linked = await manager.link_profile(connector_id, grant, completed.user)
    except Exception as failure:
        logger.warning("Failed to link an OAuth profile (%s)", type(failure).__name__)
        return page(
            title,
            "The connection is ready, but profile linking could not be completed. "
            "Ask to link your profile in the channel's chat.",
        )
    return page(
        title,
        (
            "Your channel profile is linked. "
            "You can close this tab and go back to the chat."
            if linked
            else "You can close this tab and go back to the chat."
        ),
    )

Lark / Feishu

octomate.tentacles.lark.oauth

Feishu MCP authentication and user identity verification.

LarkMcpBearerAuth

Bases: McpBearerAuth

The special case Feishu makes ordinary bearer authentication need.

Feishu's developer-app MCP endpoint carries the user token in X-Lark-MCP-UAT instead of Authorization: Bearer .... Renaming the header adds no protection against a caller holding the token. It does force an otherwise generic OAuth client to grow a provider-specific adapter.

X-Lark-MCP-Allowed-Tools carries the comma-separated mcp_tools list. Feishu's contract uses it to filter discovery and calls; omitting it exposes no tools. That makes it useful for selecting tools. Calling the header "Allowed" does not turn a caller's wish list into authorization: a client or agent holding the token and controlling HTTP requests can submit a wider list with the same credential, including on a new connection. This design adds integration work without establishing a separate permission boundary.

Octomate sets these headers from deployment configuration; proxied tool arguments cannot override them. The limitation concerns clients that hold the credential themselves. Keep grants narrow using OAuth scopes and the provider's app and resource permissions. Changing this header grants none of those permissions and does not restrict the token outside this client.

Source code in octomate/tentacles/lark/oauth.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class LarkMcpBearerAuth(McpBearerAuth):
    """The special case Feishu makes ordinary bearer authentication need.

    Feishu's developer-app MCP endpoint carries the user token in
    `X-Lark-MCP-UAT` instead of `Authorization: Bearer ...`. Renaming the header
    adds no protection against a caller holding the token. It does force an
    otherwise generic OAuth client to grow a provider-specific adapter.

    `X-Lark-MCP-Allowed-Tools` carries the comma-separated `mcp_tools` list.
    [Feishu's contract](https://open.feishu.cn/document/mcp_open_tools/developers-call-remote-mcp-server)
    uses it to filter discovery and calls; omitting it exposes no tools.
    That makes it useful for selecting tools. Calling the header "Allowed"
    does not turn a caller's wish list into authorization: a client or agent
    holding the token and controlling HTTP requests can submit a wider list
    with the same credential, including on a new connection. This design adds
    integration work without establishing a separate permission boundary.

    Octomate sets these headers from deployment configuration; proxied tool
    arguments cannot override them. The limitation concerns clients that hold
    the credential themselves. Keep grants narrow using OAuth scopes and the
    provider's app and resource permissions. Changing this header grants none
    of those permissions and does not restrict the token outside this client.
    """

    tools: list[str]

    def __init__(
        self,
        *,
        manager: OAuthManager,
        user: User,
        mcp_id: UUID7,
        url: str,
        tentacle: LarkTentacle,
    ) -> None:
        super().__init__(
            manager=manager,
            user=user,
            mcp_id=mcp_id,
            connector_id=tentacle.id,
            url=url,
        )
        self.tools = tentacle.config.mcp_tools

    def headers(self, token: SecretStr) -> dict[str, str]:
        # A caller-written allowlist is no authority over the token holder.
        # Keep this proprietary tool filter confined to the Feishu adapter.
        return {
            "X-Lark-MCP-UAT": token.get_secret_value(),
            "X-Lark-MCP-Allowed-Tools": ",".join(self.tools),
        }

tools instance-attribute 

tools: list[str] = tentacle.config.mcp_tools

headers 

headers(token: SecretStr) -> dict[str, str]
Source code in octomate/tentacles/lark/oauth.py 
65
66
67
68
69
70
71
def headers(self, token: SecretStr) -> dict[str, str]:
    # A caller-written allowlist is no authority over the token holder.
    # Keep this proprietary tool filter confined to the Feishu adapter.
    return {
        "X-Lark-MCP-UAT": token.get_secret_value(),
        "X-Lark-MCP-Allowed-Tools": ",".join(self.tools),
    }

LarkOAuthIdentity pydantic-model

Bases: BaseModel

The app-scoped user identified by Feishu's user_info endpoint.

Fields:

Source code in octomate/tentacles/lark/oauth.py 
74
75
76
77
78
class LarkOAuthIdentity(BaseModel):
    """The app-scoped user identified by Feishu's user_info endpoint."""

    open_id: str = Field(min_length=1)
    name: str

open_id pydantic-field 

open_id: str

name pydantic-field 

name: str

LarkUserInfoResponse pydantic-model

Bases: BaseModel

A successful user_info response; provider errors fail validation.

Fields:

Source code in octomate/tentacles/lark/oauth.py 
81
82
83
84
85
class LarkUserInfoResponse(BaseModel):
    """A successful user_info response; provider errors fail validation."""

    code: Literal[0]
    data: LarkOAuthIdentity

code pydantic-field 

code: Literal[0]

data pydantic-field 

data: LarkOAuthIdentity

LarkOAuthConnector pydantic-model

Bases: OAuthConnector

Link the user proven by the app's token, using its app-scoped open ID.

Feishu's API versions are a compatibility trap. Its token migration advice recommends v3, while the code_challenge parameter in Get authorization code explicitly recommends pairing PKCE with the v2 token endpoint and says support for the latest token endpoint will follow. Those recommendations contradict each other; choosing the highest version breaks this flow.

Octomate uses v1 authorization with S256 PKCE, v2 token exchange and refresh, and v1 user-info lookup. These version numbers do not describe one shared protocol version. In this integration, v3 rejected the authorization code despite the saved verifier matching the authorization challenge; switching to v2 completed a fresh authorization successfully. Keep PKCE enabled and retain the documented v1/v2 pairing until Feishu supports and verifies its replacement.

App credentials are sent in the token request's form body. The MCP headers X-Lark-MCP-UAT and X-Lark-MCP-Allowed-Tools are independent of these OAuth endpoint versions.

Fields:

Validators:

Source code in octomate/tentacles/lark/oauth.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class LarkOAuthConnector(OAuthConnector):
    """Link the user proven by the app's token, using its app-scoped open ID.

    Feishu's API versions are a compatibility trap. Its
    [token migration advice](https://open.feishu.cn/document/authentication-management/access-token/get-user-access-token)
    recommends v3, while the `code_challenge` parameter in
    [Get authorization code](https://open.feishu.cn/document/common-capabilities/sso/api/obtain-oauth-code)
    explicitly recommends pairing PKCE with the **v2 token endpoint** and says
    support for the latest token endpoint will follow. Those recommendations
    contradict each other; choosing the highest version breaks this flow.

    Octomate uses v1 authorization with S256 PKCE, v2 token exchange and refresh,
    and v1 user-info lookup. These version numbers do not describe one shared
    protocol version. In this integration, v3 rejected the authorization code
    despite the saved verifier matching the authorization challenge; switching
    to v2 completed a fresh authorization successfully. Keep PKCE enabled and
    retain the documented v1/v2 pairing until Feishu supports and verifies its
    replacement.

    App credentials are sent in the token request's form body. The MCP headers
    `X-Lark-MCP-UAT` and `X-Lark-MCP-Allowed-Tools` are independent of these OAuth
    endpoint versions.
    """

    httpx_client_factory: Callable[[], httpx2.AsyncClient]

    async def resolve_profile(self, grant: OAuthGrant) -> UserProfile:
        async with self.httpx_client_factory() as client:
            response = await OAuthTokenExchange.send(
                client,
                client.build_request(
                    "GET",
                    "https://open.feishu.cn/open-apis/authen/v1/user_info",
                    headers={
                        "Authorization": f"Bearer {grant.access_token.get_secret_value()}"
                    },
                ),
            )
            response.raise_for_status()
        identity = LarkUserInfoResponse.model_validate_json(response.content).data
        return UserProfile(channel_user_id=identity.open_id, name=identity.name)

httpx_client_factory pydantic-field 

httpx_client_factory: Callable[[], AsyncClient]

resolve_profile async 

resolve_profile(grant: OAuthGrant) -> UserProfile
Source code in octomate/tentacles/lark/oauth.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
async def resolve_profile(self, grant: OAuthGrant) -> UserProfile:
    async with self.httpx_client_factory() as client:
        response = await OAuthTokenExchange.send(
            client,
            client.build_request(
                "GET",
                "https://open.feishu.cn/open-apis/authen/v1/user_info",
                headers={
                    "Authorization": f"Bearer {grant.access_token.get_secret_value()}"
                },
            ),
        )
        response.raise_for_status()
    identity = LarkUserInfoResponse.model_validate_json(response.content).data
    return UserProfile(channel_user_id=identity.open_id, name=identity.name)