"""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 jsonutil as 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:
"""Mint an OAuth authorization link the user must visit to connect a service.
Backs the ``connect_service`` tool. Validates that the requested provider is
configured on this instance, then asks the OAuth manager to generate a
ready-to-click authorization URL for the calling user and returns a JSON
payload describing what to do next.
It fetches the process-wide OAuth manager via
``oauth_manager.get_oauth_manager()``, gates on
``OAuthManager.is_provider_configured`` (whether client credentials exist),
and delegates to ``OAuthManager.generate_connect_url`` which mints a one-time
link code and writes a JSON state payload to the Redis key
``stargazer:oauth_state:{state}`` (with a TTL) using ``ctx.redis`` before
returning the provider authorize URL; the human-readable provider blurb comes
from the module-level ``PROVIDER_INFO`` map. No model, event bus, or HTTP call
is made directly here.
Called by the tool-execution machinery (the inference worker), not by other
Python code: ``tool_loader.load_tools`` registers this function as the handler
for the ``connect_service`` entry in ``TOOLS``, and it is dispatched by name
with the schema arguments plus an injected ``ctx``. No internal call sites
were found.
Args:
provider (str): Service to connect (``github``, ``google``, ``discord``,
or ``microsoft``).
scopes (list[str] | None): Optional OAuth scopes to request; provider
defaults are used when ``None``.
ctx (ToolContext | None): Tool context supplying ``redis`` and
``user_id``; if ``None`` or lacking Redis the call short-circuits.
Returns:
str: A JSON string. On success it carries ``status`` =
``authorization_required`` with the ``provider``, ``description``,
``authorization_url``, and ``instructions``; otherwise an ``error``
message (Redis unavailable, provider not configured, or URL generation
failure).
"""
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:
"""Report whether the user has an active OAuth connection for one provider.
Backs the ``check_service_connection`` tool. Returns the connection status
for a single provider, including its stored scopes and expiry when present,
so the model can decide whether to prompt the user to connect.
It obtains the OAuth manager via ``oauth_manager.get_oauth_manager()`` and
first checks ``OAuthManager.is_provider_configured``; if a token exists it
calls ``OAuthManager.has_token`` (a non-refreshing token lookup against
``ctx.redis``) and then ``OAuthManager.list_user_connections`` to pull the
matching connection record (scopes, ``expires_at``, ``has_refresh_token``)
which is merged into the response. These reads go to the user's stored token
in Redis; no provider HTTP call is made.
Called by the tool-execution machinery (the inference worker) via the
``check_service_connection`` registration in ``TOOLS`` /
``tool_loader.load_tools``; it is dispatched by name with an injected
``ctx``. No internal Python call sites were found.
Args:
provider (str): Service to check (``github``, ``google``, ``discord``,
or ``microsoft``).
ctx (ToolContext | None): Tool context supplying ``redis`` and
``user_id``; if ``None`` or lacking Redis the call short-circuits.
Returns:
str: A JSON string with ``connected`` (bool) and ``provider``. When
connected, the matching connection fields (e.g. ``scopes``,
``expires_at``) are merged in; when not configured, ``reason`` is
``provider_not_configured``; when Redis is unavailable, an ``error``
message is returned instead.
"""
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:
"""Revoke and delete the user's stored OAuth token for one provider.
Backs the ``disconnect_service`` tool. If the user has a token for the
provider it is revoked and removed so the user must re-authorize later;
if there is nothing to disconnect the function reports that without error.
It obtains the OAuth manager via ``oauth_manager.get_oauth_manager()``,
checks ``OAuthManager.has_token`` against ``ctx.redis``, and on a hit calls
``OAuthManager.delete_token`` which best-effort revokes the token at the
provider's ``revoke_url`` (Google/Discord; failures are logged and ignored)
and then deletes the local copy from Redis. The deletion is the durable side
effect; remote revocation is best-effort.
Called by the tool-execution machinery (the inference worker) via the
``disconnect_service`` registration in ``TOOLS`` /
``tool_loader.load_tools``; it is dispatched by name with an injected
``ctx``. No internal Python call sites were found.
Args:
provider (str): Service to disconnect (``github``, ``google``,
``discord``, or ``microsoft``).
ctx (ToolContext | None): Tool context supplying ``redis`` and
``user_id``; if ``None`` or lacking Redis the call short-circuits.
Returns:
str: A JSON string with the ``provider`` and a ``status`` of
``disconnected`` when a token was removed, ``not_connected`` when there
was nothing to remove, or an ``error`` message when Redis is unavailable.
"""
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:
"""List the user's connected OAuth services and the still-available ones.
Backs the ``list_connected_services`` tool. Returns every provider the user
has connected (with scopes and expiry) alongside the configured providers
they have not yet connected, giving the model a full picture of OAuth state.
It obtains the OAuth manager via ``oauth_manager.get_oauth_manager()`` and
reads the user's connections with ``OAuthManager.list_user_connections``
(which loads each provider's stored token from ``ctx.redis``) and the set of
instances configured with credentials via
``OAuthManager.list_configured_providers``; configured providers not already
connected are surfaced as ``available_services`` using the ``PROVIDER_INFO``
blurbs. No provider HTTP call is made.
Called by the tool-execution machinery (the inference worker) via the
``list_connected_services`` registration in ``TOOLS`` /
``tool_loader.load_tools``; it is dispatched by name with an injected
``ctx``. No internal Python call sites were found.
Args:
ctx (ToolContext | None): Tool context supplying ``redis`` and
``user_id``; if ``None`` or lacking Redis the call short-circuits.
Returns:
str: A JSON string with ``connected_services`` (the user's connection
records) and ``available_services`` (configured-but-unconnected
providers with descriptions), or an ``error`` message when Redis is
unavailable.
"""
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]