"""OAuth service connection management.
Lets users connect, disconnect, and check their OAuth service connections
(GitHub, Google, Discord, Microsoft) so other tools can act on their behalf.
"""
from __future__ import annotations
import json
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
PROVIDER_INFO = {
"github": "GitHub -- repos, issues, PRs, gists, notifications",
"google": "Google -- Drive, Gmail, Calendar",
"discord": "Discord -- profile, guilds, connections",
"microsoft": "Microsoft -- OneDrive, Outlook, Calendar",
}
TOOLS = [
{
"name": "connect_service",
"description": (
"Generate an OAuth authorization link so the user can connect an external service "
"(github, google, discord, microsoft). Returns a URL the user must visit to authorize. "
"Optionally specify custom scopes."
),
"parameters": {
"type": "object",
"properties": {
"provider": {
"type": "string",
"description": "Service to connect: github, google, discord, or microsoft",
"enum": ["github", "google", "discord", "microsoft"],
},
"scopes": {
"type": "array",
"items": {"type": "string"},
"description": "Optional list of OAuth scopes to request. Uses provider defaults if omitted.",
},
},
"required": ["provider"],
},
"handler": "handle_connect_service",
},
{
"name": "check_service_connection",
"description": (
"Check if the user has connected a specific OAuth service. "
"Returns connection status and scope information."
),
"parameters": {
"type": "object",
"properties": {
"provider": {
"type": "string",
"description": "Service to check: github, google, discord, or microsoft",
"enum": ["github", "google", "discord", "microsoft"],
},
},
"required": ["provider"],
},
"handler": "handle_check_connection",
},
{
"name": "disconnect_service",
"description": (
"Disconnect (revoke) the user's OAuth tokens for a service. "
"The user will need to re-authorize to use that service again."
),
"parameters": {
"type": "object",
"properties": {
"provider": {
"type": "string",
"description": "Service to disconnect: github, google, discord, or microsoft",
"enum": ["github", "google", "discord", "microsoft"],
},
},
"required": ["provider"],
},
"handler": "handle_disconnect_service",
},
{
"name": "list_connected_services",
"description": (
"List all OAuth services the user has connected, along with their scopes "
"and expiration info. Also shows available (but unconnected) providers."
),
"parameters": {
"type": "object",
"properties": {},
},
"handler": "handle_list_connected",
},
]
# Point handler strings to actual functions
_handlers: dict[str, object] = {}
[docs]
async def handle_connect_service(provider: str, scopes: list[str] | None = None, ctx: ToolContext | None = None) -> str:
if ctx is None or ctx.redis is None:
return json.dumps({"error": "Redis not available"})
from oauth_manager import get_oauth_manager
mgr = get_oauth_manager()
if not mgr.is_provider_configured(provider):
return json.dumps({
"error": f"OAuth provider '{provider}' is not configured on this instance. "
"Ask the bot administrator to add client credentials.",
})
try:
url = await mgr.generate_connect_url(ctx.user_id, provider, ctx.redis, scopes=scopes)
except Exception as exc:
return json.dumps({"error": f"Failed to generate authorization URL: {exc}"})
desc = PROVIDER_INFO.get(provider, provider)
return json.dumps({
"status": "authorization_required",
"provider": provider,
"description": desc,
"authorization_url": url,
"instructions": f"Click the link to authorize {provider}: {url}",
})
[docs]
async def handle_check_connection(provider: str, ctx: ToolContext | None = None) -> str:
if ctx is None or ctx.redis is None:
return json.dumps({"error": "Redis not available"})
from oauth_manager import get_oauth_manager
mgr = get_oauth_manager()
if not mgr.is_provider_configured(provider):
return json.dumps({"connected": False, "reason": "provider_not_configured"})
connected = await mgr.has_token(ctx.user_id, provider, ctx.redis)
if connected:
connections = await mgr.list_user_connections(ctx.user_id, ctx.redis)
info = next((c for c in connections if c["provider"] == provider), {})
return json.dumps({"connected": True, "provider": provider, **info})
return json.dumps({"connected": False, "provider": provider})
[docs]
async def handle_disconnect_service(provider: str, ctx: ToolContext | None = None) -> str:
if ctx is None or ctx.redis is None:
return json.dumps({"error": "Redis not available"})
from oauth_manager import get_oauth_manager
mgr = get_oauth_manager()
had_token = await mgr.has_token(ctx.user_id, provider, ctx.redis)
if not had_token:
return json.dumps({"status": "not_connected", "provider": provider})
await mgr.delete_token(ctx.redis, ctx.user_id, provider)
return json.dumps({"status": "disconnected", "provider": provider})
[docs]
async def handle_list_connected(ctx: ToolContext | None = None) -> str:
if ctx is None or ctx.redis is None:
return json.dumps({"error": "Redis not available"})
from oauth_manager import get_oauth_manager
mgr = get_oauth_manager()
connections = await mgr.list_user_connections(ctx.user_id, ctx.redis)
configured = mgr.list_configured_providers()
connected_names = {c["provider"] for c in connections}
available = [
{"provider": p, "description": PROVIDER_INFO.get(p, p)}
for p in configured
if p not in connected_names
]
return json.dumps({
"connected_services": connections,
"available_services": available,
})
# Resolve handler strings to functions
for tool in TOOLS:
handler_name = tool.pop("handler")
tool["handler"] = globals()[handler_name]