mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-02 15:32:04 +00:00
f5bfa2c0e3
* refactor mcp extension and expand capabilities * update test * test file name * type definition * missing docstring * remove logger.info item not needed * readme typo * reload not available, bad example * another reload * add server prompts * update readme with server-prompts * readme update * more readme * add basic auth * improve docstring in create_mcp_server * linter * No `Dict` import * rename local * remove a debugging raise statement * Add py.typed marker --------- Co-authored-by: Theodore Aptekarev <aptekarev@gmail.com>
52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
"""Custom authentication for the MCP server."""
|
|
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import HTTPException
|
|
from fastmcp.server.auth.auth import AuthProvider
|
|
from openbb_core.api.auth.user import get_user_service
|
|
from starlette.requests import Request
|
|
|
|
|
|
class TokenAuthProvider(AuthProvider):
|
|
"""Token authentication provider."""
|
|
|
|
def __init__(self, user_service: Any):
|
|
"""Initialize the token auth provider."""
|
|
self.user_service = user_service
|
|
|
|
async def authorize(self, request: Request) -> bool:
|
|
"""Authorize the request."""
|
|
if "Authorization" not in request.headers:
|
|
return False
|
|
|
|
auth_header = request.headers["Authorization"]
|
|
try:
|
|
scheme, token = auth_header.split()
|
|
if scheme.lower() != "bearer":
|
|
return False
|
|
user = self.user_service.get_user_from_token(token)
|
|
if not user:
|
|
return False
|
|
request.state.user = user
|
|
except (ValueError, HTTPException):
|
|
return False
|
|
|
|
return True
|
|
|
|
async def verify_token(self, token: str) -> Optional[dict[str, Any]]:
|
|
"""Verify the token."""
|
|
try:
|
|
user = self.user_service.get_user_from_token(token)
|
|
if not user:
|
|
return None
|
|
return user.model_dump()
|
|
except (ValueError, HTTPException):
|
|
return None
|
|
|
|
|
|
def get_auth_provider() -> TokenAuthProvider:
|
|
"""Get the authentication provider."""
|
|
user_service = get_user_service()
|
|
return TokenAuthProvider(user_service)
|