mirror of
https://github.com/mcp-use/mcp-use.git
synced 2026-09-14 18:06:01 +08:00
18f51eddb8
* move: session * move: client * move:config * move: exceptions * move: connectors -> client.connectors * move: task_managers -> client.task_managers * move: observability -> agents.observability * move: middleware -> client.middleware * move: managers -> agents.managers * move: auth -> client.auth * move: adapters -> agents.adapters * test: bc test * test: bc test * chore: remove relative imports * fix tests * chore: update docs * chore: deprecation warnings * chore: fix dangerous non absolute import * fix: remove legacy imports * telemetry
24 lines
678 B
Python
24 lines
678 B
Python
"""Bearer token authentication support."""
|
|
|
|
from collections.abc import Generator
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, SecretStr
|
|
|
|
from mcp_use.telemetry.telemetry import telemetry
|
|
|
|
|
|
class BearerAuth(httpx.Auth, BaseModel):
|
|
"""Bearer token authentication for HTTP requests."""
|
|
|
|
token: SecretStr
|
|
|
|
def __init__(self, **data):
|
|
super().__init__(**data)
|
|
|
|
@telemetry("auth_bearer")
|
|
def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
|
|
"""Apply bearer token authentication to the request."""
|
|
request.headers["Authorization"] = f"Bearer {self.token.get_secret_value()}"
|
|
yield request
|