Coverage for polar/oauth2/mcp_client.py: 0%

1"""Script to generate an OAuth2 client for our MCP server.""" 

2 

3import argparse 

4import asyncio 

5from pathlib import Path 

6 

7from rich import print 

8 

9from polar.auth.scope import SCOPES_SUPPORTED 

10from polar.kit.crypto import generate_token 

11from polar.kit.db.postgres import create_async_sessionmaker 

12from polar.models import OAuth2Client 

13from polar.postgres import create_async_engine 

14 

15from .constants import ( 

16 CLIENT_ID_PREFIX, 

17 CLIENT_REGISTRATION_TOKEN_PREFIX, 

18 CLIENT_SECRET_PREFIX, 

19) 

20 

21 

22async def create_client(add_to_env_file: bool) -> None: 

23 engine = create_async_engine("script") 

24 sessionmaker = create_async_sessionmaker(engine) 

25 async with sessionmaker() as session: 

26 oauth2_client = OAuth2Client( 

27 client_id=generate_token(prefix=CLIENT_ID_PREFIX), 

28 client_secret=generate_token(prefix=CLIENT_SECRET_PREFIX), 

29 registration_access_token=generate_token( 

30 prefix=CLIENT_REGISTRATION_TOKEN_PREFIX 

31 ), 

32 user=None, 

33 ) 

34 oauth2_client.set_client_metadata( 

35 { 

36 "client_name": "Polar MCP Client", 

37 "redirect_uris": [], 

38 "token_endpoint_auth_method": "client_secret_post", 

39 "grant_types": ["web"], 

40 "response_types": [], 

41 "scope": " ".join(SCOPES_SUPPORTED), 

42 } 

43 ) 

44 session.add(oauth2_client) 

45 await session.commit() 

46 

47 print("[bold green]OAuth2 Client created successfully![/bold green]") 

48 

49 if add_to_env_file: 

50 env_file_path = ( 

51 Path(__file__).parent.parent.parent.parent 

52 / "clients" 

53 / "apps" 

54 / "web" 

55 / ".env.local" 

56 ) 

57 with open(env_file_path, "a") as f: 

58 f.write(f"\nMCP_OAUTH2_CLIENT_ID={oauth2_client.client_id}\n") 

59 f.write(f"MCP_OAUTH2_CLIENT_SECRET={oauth2_client.client_secret}\n") 

60 print(f"[bold blue]Credentials added to {env_file_path}[/bold blue]") 

61 else: 

62 print(f"Client ID: [bold]{oauth2_client.client_id}[/bold]") 

63 print(f"Client Secret: [bold]{oauth2_client.client_secret}[/bold]") 

64 

65 

66if __name__ == "__main__": 

67 parser = argparse.ArgumentParser( 

68 description="Generate an OAuth2 client for MCP server" 

69 ) 

70 parser.add_argument( 

71 "--add-to-env-file", 

72 action="store_true", 

73 help="Add credentials to .env.local file instead of printing them", 

74 ) 

75 

76 args = parser.parse_args() 

77 asyncio.run(create_client(args.add_to_env_file))