# Directory Structure
```
├── Dockerfile
├── google_chat.py
├── LICENSE
├── poetry.lock
├── pyproject.toml
├── README.md
├── requirements.txt
├── server_auth.py
├── server.py
├── src
│ └── mcp_gcp_chat_py
│ └── __init__.py
├── tests
│ └── __init__.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Introduction
2 |
3 | This project provides a Google Chat integration for MCP (Model Control Protocol) servers written by Python with FastMCP. It allows you to access and interact with Google Chat spaces and messages through MCP tools.
4 |
5 | ## Structure
6 | The project consists of two main components:
7 |
8 | 1. **MCP Server with Google Chat Tools**: Provides tools for interacting with Google Chat through the Model Control Protocol.
9 | - Written by FastMCP
10 | - `server.py`: Main MCP server implementation with Google Chat tools
11 | - `google_chat.py`: Google Chat API integration and authentication handling
12 |
13 | 2. **Authentication Server**: Standalone component for Google account authentication
14 | - Written by FastAPI
15 | - Handles OAuth2 flow with Google
16 | - Stores and manages access tokens
17 | - Can be run independently or as part of the MCP server
18 | - `server_auth.py`: Authentication server implementation
19 |
20 | The authentication flow allows you to obtain and refresh Google API tokens, which are then used by the MCP tools to access Google Chat data. (Your spaces and messages)
21 |
22 |
23 | ## Features
24 |
25 | - OAuth2 authentication with Google Chat API
26 | - List available Google Chat spaces
27 | - Retrieve messages from specific spaces with date filtering
28 | - Local authentication server for easy setup
29 |
30 | ## Requirements
31 |
32 | - Python 3.8+
33 | - Google Cloud project with Chat API enabled
34 | - OAuth2 credentials from Google Cloud Console
35 |
36 | # How to use?
37 |
38 | ## Prepare Google Oauth Login
39 | 1. Clone this project
40 | ```
41 | git clone https://github.com/chy168/google-chat-mcp-server.git
42 | cd google-chat-mcp-server
43 | ```
44 | 2. Prepare a Google Cloud Project (GCP)
45 | 3. Google Cloud Conolse (https://console.cloud.google.com/auth/overview?project=<YOUR_PROJECT_NAME>)
46 | 4. Google Auth Platform > Clients > (+) Create client > Web application
47 | reference: https://developers.google.com/identity/protocols/oauth2/?hl=en
48 | Authorized JavaScript origins add: `http://localhost:8000`
49 | Authorized redirect URIs: `http://localhost:8000/auth/callback`
50 | 5. After you create a OAuth 2.0 Client, download the client secrets as `.json` file. Save as `credentials.json` at top level of project.
51 |
52 |
53 | ## Run Auth server and get your Google access token (login google only, not MCP server yet)
54 | ```
55 | python server.py -local-auth --port 8000
56 | ```
57 |
58 | - Open browser at http://localhost:8000/auth
59 | - login it!
60 | - after loggined, you access token will be saved as `token.json`
61 |
62 | ## MCP Configuration (mcp.json)
63 | ```
64 | {
65 | "mcpServers": {
66 | "google_chat": {
67 | "command": "uv",
68 | "args": [
69 | "--directory",
70 | "<YOUR_REPO_PATH>/google-chat-mcp-server",
71 | "run",
72 | "server.py",
73 | "--token-path",
74 | "<YOUR_REPO_PATH>/google-chat-mcp-server/token.json"
75 | ]
76 | }
77 | }
78 | ```
79 |
80 | ## Tools
81 | The MCP server provides the following tools:
82 |
83 | ### Google Chat Tools
84 | - `get_chat_spaces()` - List all Google Chat spaces the bot has access to
85 | - `get_space_messages(space_name: str, start_date: str, end_date: str = None)` - List messages from a specific Google Chat space with optional time filtering
86 |
87 |
88 | ## Development and Debug
89 | ```
90 | fastmcp dev server.py --with-editable .
91 | ```
92 |
93 |
```
--------------------------------------------------------------------------------
/src/mcp_gcp_chat_py/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | fastmcp>=0.1.0
2 | httpx>=0.24.0
3 | google-api-core>=2.11.0
4 | google-api-python-client>=2.0.0
5 | google-auth>=2.22.0
6 | google-auth-httplib2>=0.1.0
7 | google-auth-oauthlib>=1.0.0
8 | fastapi>=0.68.0
9 | uvicorn>=0.15.0
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "mcp-gcp-chat-py"
3 | version = "0.1.0"
4 | description = ""
5 | authors = [
6 | {name = "Zz Chen",email = "[email protected]"}
7 | ]
8 | readme = "README.md"
9 | requires-python = ">=3.11"
10 | dependencies = [
11 | "fastmcp (>=0.4.1,<0.5.0)"
12 | ]
13 |
14 | [tool.poetry]
15 | packages = [{include = "mcp_gcp_chat_py", from = "src"}]
16 |
17 |
18 | [build-system]
19 | requires = ["poetry-core>=2.0.0,<3.0.0"]
20 | build-backend = "poetry.core.masonry.api"
21 |
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
1 | # Use official Python image
2 | FROM python:3.11-slim
3 |
4 | # Set work directory
5 | WORKDIR /app
6 |
7 | # Install system dependencies
8 | RUN apt-get update && apt-get install -y --no-install-recommends \
9 | build-essential \
10 | && rm -rf /var/lib/apt/lists/*
11 |
12 | # Copy project files
13 | COPY . /app
14 |
15 | # Install Poetry
16 | RUN pip install poetry
17 |
18 | # Install dependencies
19 | RUN poetry config virtualenvs.create false \
20 | && poetry install --no-interaction --no-ansi --only main
21 |
22 | # Expose MCP default port
23 | EXPOSE 8080
24 |
25 | # Set environment variables (if needed)
26 | ENV PYTHONUNBUFFERED=1
27 |
28 | # Start the MCP server (adjust if entrypoint is different)
29 | CMD ["python", "server.py"]
30 |
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
1 | # server.py
2 | import httpx
3 | import sys
4 | import argparse
5 | from typing import List, Dict
6 |
7 | from fastmcp import FastMCP
8 | from google_chat import list_chat_spaces, DEFAULT_CALLBACK_URL, set_token_path
9 | from server_auth import run_auth_server
10 |
11 | # Create an MCP server
12 | mcp = FastMCP("Demo")
13 |
14 | # Add an addition tool
15 | @mcp.tool()
16 | def add(a: int, b: int) -> int:
17 | """Add two numbers"""
18 | return a + b
19 |
20 | name = "GG"
21 |
22 | # Add a dynamic greeting resource
23 | @mcp.resource("greeting://{name}")
24 | def get_greeting(name: str) -> str:
25 | """Get a personalized greeting"""
26 | return f"Hello, {name}!"
27 |
28 | @mcp.tool()
29 | async def fetch_weather(city: str) -> str:
30 | """Fetch current weather for a city"""
31 | async with httpx.AsyncClient() as client:
32 | response = await client.get(
33 | f"https://api.weather.com/{city}"
34 | )
35 | return response.text
36 |
37 | @mcp.tool()
38 | async def get_ip_my_address(city: str) -> str:
39 | """Get IP address from outian.net"""
40 | async with httpx.AsyncClient() as client:
41 | response = await client.get(
42 | f"http://outian.net/"
43 | )
44 | return response.text
45 |
46 | @mcp.tool()
47 | async def get_chat_spaces() -> List[Dict]:
48 | """List all Google Chat spaces the bot has access to.
49 |
50 | This tool requires OAuth authentication. On first run, it will open a browser window
51 | for you to log in with your Google account. Make sure you have credentials.json
52 | downloaded from Google Cloud Console in the current directory.
53 | """
54 | return await list_chat_spaces()
55 |
56 | @mcp.tool()
57 | async def get_space_messages(space_name: str,
58 | start_date: str,
59 | end_date: str = None) -> List[Dict]:
60 | """List messages from a specific Google Chat space with optional time filtering.
61 |
62 | This tool requires OAuth authentication. The space_name should be in the format
63 | 'spaces/your_space_id'. Dates should be in YYYY-MM-DD format (e.g., '2024-03-22').
64 |
65 | When only start_date is provided, it will query messages for that entire day.
66 | When both dates are provided, it will query messages from start_date 00:00:00Z
67 | to end_date 23:59:59Z.
68 |
69 | Args:
70 | space_name: The name/identifier of the space to fetch messages from
71 | start_date: Required start date in YYYY-MM-DD format
72 | end_date: Optional end date in YYYY-MM-DD format
73 |
74 | Returns:
75 | List of message objects from the space matching the time criteria
76 |
77 | Raises:
78 | ValueError: If the date format is invalid or dates are in wrong order
79 | """
80 | from google_chat import list_space_messages
81 | from datetime import datetime, timezone
82 |
83 | try:
84 | # Parse start date and set to beginning of day (00:00:00Z)
85 | start_datetime = datetime.strptime(start_date, '%Y-%m-%d').replace(
86 | hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc
87 | )
88 |
89 | # Parse end date if provided and set to end of day (23:59:59Z)
90 | end_datetime = None
91 | if end_date:
92 | end_datetime = datetime.strptime(end_date, '%Y-%m-%d').replace(
93 | hour=23, minute=59, second=59, microsecond=999999, tzinfo=timezone.utc
94 | )
95 |
96 | # Validate date range
97 | if start_datetime > end_datetime:
98 | raise ValueError("start_date must be before end_date")
99 | except ValueError as e:
100 | if "strptime" in str(e):
101 | raise ValueError("Dates must be in YYYY-MM-DD format (e.g., '2024-03-22')")
102 | raise e
103 |
104 | return await list_space_messages(space_name, start_datetime, end_datetime)
105 |
106 | if __name__ == "__main__":
107 | parser = argparse.ArgumentParser(description='MCP Server with Google Chat Authentication')
108 | parser.add_argument('-local-auth', action='store_true', help='Run the local authentication server')
109 | parser.add_argument('--host', default='localhost', help='Host to bind the server to (default: localhost)')
110 | parser.add_argument('--port', type=int, default=8000, help='Port to run the server on (default: 8000)')
111 | parser.add_argument('--token-path', default='token.json', help='Path to store OAuth token (default: token.json)')
112 |
113 | args = parser.parse_args()
114 |
115 | # Set the token path for OAuth storage
116 | set_token_path(args.token_path)
117 |
118 | if args.local_auth:
119 | print(f"\nStarting local authentication server at http://{args.host}:{args.port}")
120 | print("Available endpoints:")
121 | print(" - /auth : Start OAuth authentication flow")
122 | print(" - /status : Check authentication status")
123 | print(" - /auth/callback : OAuth callback endpoint")
124 | print(f"\nDefault callback URL: {DEFAULT_CALLBACK_URL}")
125 | print(f"Token will be stored at: {args.token_path}")
126 | print("\nPress CTRL+C to stop the server")
127 | print("-" * 50)
128 | run_auth_server(port=args.port, host=args.host)
129 | else:
130 | mcp.run()
```
--------------------------------------------------------------------------------
/google_chat.py:
--------------------------------------------------------------------------------
```python
1 | import os
2 | import json
3 | import datetime
4 | from typing import List, Dict, Optional, Tuple
5 | from google.oauth2.credentials import Credentials
6 | from google_auth_oauthlib.flow import InstalledAppFlow
7 | from google.auth.transport.requests import Request
8 | from googleapiclient.discovery import build
9 | from pathlib import Path
10 |
11 | # If modifying these scopes, delete the file token.json.
12 | SCOPES = [
13 | 'https://www.googleapis.com/auth/chat.spaces.readonly',
14 | 'https://www.googleapis.com/auth/chat.messages'
15 | ]
16 | DEFAULT_CALLBACK_URL = "http://localhost:8000/auth/callback"
17 | DEFAULT_TOKEN_PATH = 'token.json'
18 |
19 | # Store credentials info
20 | token_info = {
21 | 'credentials': None,
22 | 'last_refresh': None,
23 | 'token_path': DEFAULT_TOKEN_PATH
24 | }
25 |
26 | def set_token_path(path: str) -> None:
27 | """Set the global token path for OAuth storage.
28 |
29 | Args:
30 | path: Path where the token should be stored
31 | """
32 | token_info['token_path'] = path
33 |
34 | def save_credentials(creds: Credentials, token_path: Optional[str] = None) -> None:
35 | """Save credentials to file and update in-memory cache.
36 |
37 | Args:
38 | creds: The credentials to save
39 | token_path: Path to save the token file
40 | """
41 | # Use configured token path if none provided
42 | if token_path is None:
43 | token_path = token_info['token_path']
44 |
45 | # Save to file
46 | token_path = Path(token_path)
47 | with open(token_path, 'w') as token:
48 | token.write(creds.to_json())
49 |
50 | # Update in-memory cache
51 | token_info['credentials'] = creds
52 | token_info['last_refresh'] = datetime.datetime.utcnow()
53 |
54 | def get_credentials(token_path: Optional[str] = None) -> Optional[Credentials]:
55 | """Gets valid user credentials from storage or memory.
56 |
57 | Args:
58 | token_path: Optional path to token file. If None, uses the configured path.
59 |
60 | Returns:
61 | Credentials object or None if no valid credentials exist
62 | """
63 | if token_path is None:
64 | token_path = token_info['token_path']
65 |
66 | creds = token_info['credentials']
67 |
68 | # If no credentials in memory, try to load from file
69 | if not creds:
70 | token_path = Path(token_path)
71 | if token_path.exists():
72 | creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)
73 | token_info['credentials'] = creds
74 |
75 | # If we have credentials that need refresh
76 | if creds and creds.expired and creds.refresh_token:
77 | try:
78 | creds.refresh(Request())
79 | save_credentials(creds, token_path)
80 | except Exception:
81 | return None
82 |
83 | return creds if (creds and creds.valid) else None
84 |
85 | async def refresh_token(token_path: Optional[str] = None) -> Tuple[bool, str]:
86 | """Attempt to refresh the current token.
87 |
88 | Args:
89 | token_path: Path to the token file. If None, uses the configured path.
90 |
91 | Returns:
92 | Tuple of (success: bool, message: str)
93 | """
94 | if token_path is None:
95 | token_path = token_info['token_path']
96 |
97 | try:
98 | creds = token_info['credentials']
99 | if not creds:
100 | token_path = Path(token_path)
101 | if not token_path.exists():
102 | return False, "No token file found"
103 | creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)
104 |
105 | if not creds.refresh_token:
106 | return False, "No refresh token available"
107 |
108 | creds.refresh(Request())
109 | save_credentials(creds, token_path)
110 | return True, "Token refreshed successfully"
111 | except Exception as e:
112 | return False, f"Failed to refresh token: {str(e)}"
113 |
114 | # MCP functions
115 | async def list_chat_spaces() -> List[Dict]:
116 | """Lists all Google Chat spaces the bot has access to."""
117 | try:
118 | creds = get_credentials()
119 | if not creds:
120 | raise Exception("No valid credentials found. Please authenticate first.")
121 |
122 | service = build('chat', 'v1', credentials=creds)
123 | spaces = service.spaces().list(pageSize=30).execute()
124 | return spaces.get('spaces', [])
125 | except Exception as e:
126 | raise Exception(f"Failed to list chat spaces: {str(e)}")
127 |
128 | async def list_space_messages(space_name: str,
129 | start_date: Optional[datetime.datetime] = None,
130 | end_date: Optional[datetime.datetime] = None) -> List[Dict]:
131 | """Lists messages from a specific Google Chat space with optional time filtering.
132 |
133 | Args:
134 | space_name: The name/identifier of the space to fetch messages from
135 | start_date: Optional start datetime for filtering messages. If provided without end_date,
136 | will query messages for the entire day of start_date
137 | end_date: Optional end datetime for filtering messages. Only used if start_date is also provided
138 |
139 | Returns:
140 | List of message objects from the space matching the time criteria
141 |
142 | Raises:
143 | Exception: If authentication fails or API request fails
144 | """
145 | try:
146 | creds = get_credentials()
147 | if not creds:
148 | raise Exception("No valid credentials found. Please authenticate first.")
149 |
150 | service = build('chat', 'v1', credentials=creds)
151 |
152 | # Prepare filter string based on provided dates
153 | filter_str = None
154 | if start_date:
155 | if end_date:
156 | # Format for date range query
157 | filter_str = f"createTime > \"{start_date.isoformat()}\" AND createTime < \"{end_date.isoformat()}\""
158 | else:
159 | # For single day query, set range from start of day to end of day
160 | day_start = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
161 | day_end = day_start + datetime.timedelta(days=1)
162 | filter_str = f"createTime > \"{day_start.isoformat()}\" AND createTime < \"{day_end.isoformat()}\""
163 |
164 | # Make API request
165 | request = service.spaces().messages().list(parent=space_name, pageSize=100)
166 | if filter_str:
167 | request = service.spaces().messages().list(parent=space_name, filter=filter_str, pageSize=100)
168 |
169 | response = request.execute()
170 |
171 | return response.get('messages', [])
172 |
173 | except Exception as e:
174 | raise Exception(f"Failed to list messages in space: {str(e)}")
175 |
176 |
```
--------------------------------------------------------------------------------
/server_auth.py:
--------------------------------------------------------------------------------
```python
1 | from fastapi import FastAPI, HTTPException, Query
2 | from fastapi.responses import JSONResponse, RedirectResponse
3 | import uvicorn
4 | import signal
5 | import asyncio
6 | from pathlib import Path
7 | from typing import Optional, Dict
8 | from datetime import datetime
9 | from google_auth_oauthlib.flow import InstalledAppFlow
10 |
11 | from google_chat import (
12 | get_credentials,
13 | save_credentials,
14 | refresh_token,
15 | SCOPES,
16 | DEFAULT_CALLBACK_URL,
17 | token_info
18 | )
19 |
20 | # Store OAuth flow state
21 | oauth_flows: Dict[str, InstalledAppFlow] = {}
22 |
23 | # Create FastAPI app for local auth server
24 | app = FastAPI(title="Google Chat Auth Server")
25 |
26 | @app.get("/auth")
27 | async def start_auth(callback_url: Optional[str] = Query(None)):
28 | """Start OAuth authentication flow"""
29 | try:
30 | # Check if we already have valid credentials
31 | if get_credentials():
32 | return JSONResponse(
33 | content={
34 | "status": "already_authenticated",
35 | "message": "Valid credentials already exist"
36 | }
37 | )
38 |
39 | # Initialize OAuth 2.0 flow
40 | credentials_path = Path('credentials.json')
41 | if not credentials_path.exists():
42 | raise FileNotFoundError(
43 | "credentials.json not found. Please download it from Google Cloud Console "
44 | "and save it in the current directory."
45 | )
46 |
47 | flow = InstalledAppFlow.from_client_secrets_file(
48 | str(credentials_path),
49 | SCOPES,
50 | redirect_uri=callback_url or DEFAULT_CALLBACK_URL
51 | )
52 |
53 | # Generate authorization URL with offline access and force approval
54 | auth_url, state = flow.authorization_url(
55 | access_type='offline', # Enable offline access
56 | prompt='consent', # Force consent screen to ensure refresh token
57 | include_granted_scopes='true'
58 | )
59 |
60 | # Store the flow object for later use
61 | oauth_flows[state] = flow
62 |
63 | # Redirect user to Google's auth page
64 | return RedirectResponse(url=auth_url)
65 |
66 | except Exception as e:
67 | raise HTTPException(status_code=500, detail=str(e))
68 |
69 | @app.get("/auth/callback")
70 | async def auth_callback(
71 | state: str = Query(...),
72 | code: Optional[str] = Query(None),
73 | error: Optional[str] = Query(None)
74 | ):
75 | """Handle OAuth callback"""
76 | try:
77 | if error:
78 | raise HTTPException(
79 | status_code=400,
80 | detail=f"Authorization failed: {error}"
81 | )
82 |
83 | if not code:
84 | raise HTTPException(
85 | status_code=400,
86 | detail="No authorization code received"
87 | )
88 |
89 | # Retrieve the flow object
90 | flow = oauth_flows.get(state)
91 | if not flow:
92 | raise HTTPException(
93 | status_code=400,
94 | detail="Invalid state parameter"
95 | )
96 |
97 | try:
98 | # Exchange auth code for credentials with offline access
99 | print("fetching token: ", code)
100 | flow.fetch_token(
101 | code=code,
102 | # Ensure we're requesting offline access for refresh tokens
103 | access_type='offline'
104 | )
105 | print("fetched credentials: ", flow.credentials)
106 | creds = flow.credentials
107 |
108 | # Verify we got a refresh token
109 | if not creds.refresh_token:
110 | raise HTTPException(
111 | status_code=400,
112 | detail="Failed to obtain refresh token. Please try again."
113 | )
114 |
115 | # Save credentials both to file and memory
116 | print("saving credentials: ", creds)
117 | save_credentials(creds)
118 |
119 | # Clean up the flow object
120 | del oauth_flows[state]
121 |
122 | return JSONResponse(
123 | content={
124 | "status": "success",
125 | "message": "Authorization successful. Long-lived token obtained. You can close this window.",
126 | "token_file": token_info['token_path'],
127 | "expires_at": creds.expiry.isoformat() if creds.expiry else None,
128 | "has_refresh_token": bool(creds.refresh_token)
129 | }
130 | )
131 | except Exception as e:
132 | # Clean up flow object even if there's an error
133 | del oauth_flows[state]
134 | raise
135 |
136 | except Exception as e:
137 | raise HTTPException(status_code=500, detail=str(e))
138 |
139 | @app.post("/auth/refresh")
140 | async def manual_token_refresh():
141 | """Manually trigger a token refresh"""
142 | success, message = await refresh_token()
143 | if success:
144 | creds = token_info['credentials']
145 | return JSONResponse(
146 | content={
147 | "status": "success",
148 | "message": message,
149 | "expires_at": creds.expiry.isoformat() if creds.expiry else None,
150 | "last_refresh": token_info['last_refresh'].isoformat()
151 | }
152 | )
153 | else:
154 | raise HTTPException(
155 | status_code=400,
156 | detail=message
157 | )
158 |
159 | @app.get("/status")
160 | async def check_auth_status():
161 | """Check if we have valid credentials"""
162 | token_path = token_info['token_path']
163 | token_file = Path(token_path)
164 | if not token_file.exists():
165 | return JSONResponse(
166 | content={
167 | "status": "not_authenticated",
168 | "message": "No authentication token found",
169 | "token_path": str(token_path)
170 | }
171 | )
172 |
173 | try:
174 | creds = get_credentials()
175 | if creds:
176 | return JSONResponse(
177 | content={
178 | "status": "authenticated",
179 | "message": "Valid credentials exist",
180 | "token_path": str(token_path),
181 | "expires_at": creds.expiry.isoformat() if creds.expiry else None,
182 | "last_refresh": token_info['last_refresh'].isoformat() if token_info['last_refresh'] else None,
183 | "has_refresh_token": bool(creds.refresh_token)
184 | }
185 | )
186 | else:
187 | return JSONResponse(
188 | content={
189 | "status": "expired",
190 | "message": "Credentials exist but are expired or invalid",
191 | "token_path": str(token_path)
192 | }
193 | )
194 | except Exception as e:
195 | return JSONResponse(
196 | content={
197 | "status": "error",
198 | "message": str(e),
199 | "token_path": str(token_path)
200 | },
201 | status_code=500
202 | )
203 |
204 | def run_auth_server(port: int = 8000, host: str = "localhost"):
205 | """Run the authentication server with graceful shutdown support
206 |
207 | Args:
208 | port: Port to run the server on (default: 8000)
209 | host: Host to bind the server to (default: localhost)
210 | """
211 | server_config = uvicorn.Config(app, host=host, port=port)
212 | server = uvicorn.Server(server_config)
213 |
214 | # Handle graceful shutdown
215 | def signal_handler(signum, frame):
216 | print("\nReceived signal to terminate. Performing graceful shutdown...")
217 | asyncio.create_task(server.shutdown())
218 |
219 | # Register signal handlers
220 | signal.signal(signal.SIGINT, signal_handler) # Handle Ctrl+C
221 | signal.signal(signal.SIGTERM, signal_handler) # Handle termination signal
222 |
223 | try:
224 | print(f"\nServer is running at: http://{host}:{port}")
225 | print(f"Default callback URL: {DEFAULT_CALLBACK_URL}")
226 | # Start the server
227 | server.run()
228 | except KeyboardInterrupt:
229 | print("\nShutting down the auth server...")
230 | finally:
231 | print("Auth server has been stopped.")
```