MCP Server Box

by box-community
Verified
from typing import Optional from box_sdk_gen import ( BoxClient, CCGConfig, BoxCCGAuth, FileWithInMemoryCacheTokenStorage, OAuthConfig, BoxOAuth, GetAuthorizeUrlOptions, AccessToken, ) from dotenv import load_dotenv import os import uuid from .box_auth_callback import open_browser, callback_handle_request load_dotenv() # Environment variables CLIENT_ID = os.getenv("BOX_CLIENT_ID") CLIENT_SECRET = os.getenv("BOX_CLIENT_SECRET") # CCG SUBJECT_TYPE = os.getenv("BOX_SUBJECT_TYPE") SUBJECT_ID = os.getenv("BOX_SUBJECT_ID") # OAuth REDIRECT_URL = os.environ.get("BOX_REDIRECT_URL", "http://localhost:8000/callback") def get_auth_config() -> OAuthConfig: return OAuthConfig( client_id=CLIENT_ID, client_secret=CLIENT_SECRET, token_storage=FileWithInMemoryCacheTokenStorage(".auth.oauth"), ) def get_ccg_config() -> CCGConfig: if SUBJECT_TYPE == "enterprise": enterprise_id = SUBJECT_ID user_id = None else: enterprise_id = None user_id = SUBJECT_ID return CCGConfig( client_id=CLIENT_ID, client_secret=CLIENT_SECRET, enterprise_id=enterprise_id, user_id=user_id, token_storage=FileWithInMemoryCacheTokenStorage(".auth.ccg"), ) def get_ccg_client() -> BoxClient: conf = get_ccg_config() auth = BoxCCGAuth(conf) return add_extra_header_to_box_client(BoxClient(auth)) def get_oauth_client() -> BoxClient: conf = get_auth_config() auth = BoxOAuth(conf) return add_extra_header_to_box_client(BoxClient(auth)) def add_extra_header_to_box_client(box_client: BoxClient) -> BoxClient: """ Add extra headers to the Box client. Args: box_client (BoxClient): A Box client object. header (Dict[str, str]): A dictionary of extra headers to add to the Box client. Returns: BoxClient: A Box client object with the extra headers added. """ header = {"x-box-ai-library": "mcp-server-box"} return box_client.with_extra_headers(extra_headers=header) def authorize_app() -> bool: conf = get_auth_config() auth = BoxOAuth(conf) state = str(uuid.uuid4()) options = GetAuthorizeUrlOptions( client_id=CLIENT_ID, redirect_uri=REDIRECT_URL, state=state, ) auth_url = auth.get_authorize_url(options=options) open_browser(auth_url) hostname = REDIRECT_URL.split(":")[1].replace("/", "") port = REDIRECT_URL.split(":")[2].split("/")[0] callback_handle_request(auth, hostname, int(port), state) access_token = auth.token_storage.get() if not access_token: return False return True