mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-12 22:14:05 +00:00
arc53-machine
54ba7c7b9b
The connection setup endpoint takes sync.config, validated like an upload's config, and passes it to the ingest task so the synced source gets the chosen chunking and retrieval settings. An invalid config is refused before the idempotency key is claimed.
723 lines
32 KiB
Python
723 lines
32 KiB
Python
"""Connectors catalog and connections API.
|
|
|
|
``/api/connectors/catalog`` lists every service DocsGPT can connect to, with
|
|
whether the server is set up for it and the caller's connection summary.
|
|
``/api/connections`` lists and manages the caller's connections. Responses
|
|
never include tokens or secrets.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from flask import current_app, jsonify, make_response, request
|
|
from flask_restx import Namespace, Resource
|
|
|
|
import uuid
|
|
|
|
from docsgpt.api import api
|
|
from docsgpt.api.user.authz import ROLE_ADMIN, has_role
|
|
from docsgpt.connectors import catalog, service
|
|
from docsgpt.storage.db.repositories.connector_sessions import ConnectorSessionsRepository
|
|
from docsgpt.storage.db.session import db_readonly, db_session
|
|
from docsgpt.security.encryption import CredentialDecryptionError
|
|
|
|
_FREQUENCIES = ("never", "daily", "weekly", "monthly")
|
|
|
|
connections_ns = Namespace("connections", description="Connectors and connections", path="/api")
|
|
api.add_namespace(connections_ns)
|
|
|
|
|
|
def _user_id() -> str | None:
|
|
token = getattr(request, "decoded_token", None)
|
|
return token.get("sub") if isinstance(token, dict) else None
|
|
|
|
|
|
def _unauthorized():
|
|
return make_response(jsonify({"success": False, "error": "Unauthorized"}), 401)
|
|
|
|
|
|
def _not_found():
|
|
return make_response(jsonify({"success": False, "error": "Connection not found"}), 404)
|
|
|
|
|
|
@connections_ns.route("/connectors/catalog")
|
|
class ConnectorCatalog(Resource):
|
|
@api.doc(description="Every connector with its availability and the caller's connection summary")
|
|
def get(self):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
try:
|
|
with db_readonly() as conn:
|
|
entries = service.catalog_for_user(
|
|
conn, user_id, is_admin=has_role(request.decoded_token, ROLE_ADMIN),
|
|
)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error building connector catalog: {err}", exc_info=True)
|
|
return make_response(jsonify({"success": False, "error": "Failed to load connectors"}), 500)
|
|
return make_response(jsonify({"success": True, "connectors": entries}), 200)
|
|
|
|
|
|
def _json_body() -> dict:
|
|
body = request.get_json(silent=True)
|
|
return body if isinstance(body, dict) else {}
|
|
|
|
|
|
def _owned(conn, connection_id: str, user_id: str):
|
|
return ConnectorSessionsRepository(conn).get_for_user(connection_id, user_id)
|
|
|
|
|
|
def _error(message: str, status: int, **extra):
|
|
return make_response(jsonify({"success": False, "error": message, **extra}), status)
|
|
|
|
|
|
@connections_ns.route("/connections")
|
|
class ConnectionsList(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"Create a connection from pasted credentials: "
|
|
"{connector_key, credentials, label?}. Same credentials reuse the same connection."
|
|
)
|
|
)
|
|
def post(self):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
body = _json_body()
|
|
definition = catalog.get_definition(body.get("connector_key"))
|
|
if definition is None or definition.auth_kind != "api_key":
|
|
return _error("This connector does not take pasted credentials", 400)
|
|
if definition.missing_settings:
|
|
return _error("This connector needs admin setup", 400, code="needs_setup")
|
|
credentials = body.get("credentials")
|
|
if not isinstance(credentials, dict):
|
|
return _error("credentials must be an object", 400)
|
|
label = body.get("label") or None
|
|
if definition.key == "github":
|
|
# Check the token now, not at the first sync, and name the
|
|
# connection after the account rather than a hint of the token.
|
|
from docsgpt.connectors import github
|
|
|
|
try:
|
|
label = label or github.token_account(credentials.get("access_token"))
|
|
except github.TokenRejected as err:
|
|
return _error(str(err), 400, code="invalid_credentials")
|
|
except service.TransientConnectionError as err:
|
|
return _error(str(err), 502)
|
|
try:
|
|
with db_session() as conn:
|
|
row, created = service.create_api_key_connection(
|
|
conn, user_id, definition, credentials, label=label,
|
|
)
|
|
except service.EncryptionKeyNotConfigured as err:
|
|
return _error(str(err), 400, code="encryption_key_default")
|
|
except service.ConnectorDisabled as err:
|
|
return _error(str(err), 403, code="disabled")
|
|
except ValueError as err:
|
|
return _error(str(err), 400)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error creating connection: {err}", exc_info=True)
|
|
return _error("Failed to create connection", 500)
|
|
return make_response(
|
|
jsonify(
|
|
{
|
|
"success": True,
|
|
"created": created,
|
|
"connection": service.serialize_connection(row),
|
|
"setup": dict(definition.setup),
|
|
}
|
|
),
|
|
201 if created else 200,
|
|
)
|
|
|
|
@api.doc(description="The caller's connections with status and linked resource counts")
|
|
def get(self):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
try:
|
|
with db_readonly() as conn:
|
|
connections = service.list_connections(conn, user_id)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error listing connections: {err}", exc_info=True)
|
|
return make_response(jsonify({"success": False, "error": "Failed to load connections"}), 500)
|
|
return make_response(jsonify({"success": True, "connections": connections}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>")
|
|
class ConnectionDetail(Resource):
|
|
@api.doc(description="One connection with the sources it syncs and the tools it provides")
|
|
def get(self, connection_id: str):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
try:
|
|
with db_readonly() as conn:
|
|
row = ConnectorSessionsRepository(conn).get_for_user(connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
detail = service.connection_detail(conn, row)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error loading connection: {err}", exc_info=True)
|
|
return make_response(jsonify({"success": False, "error": "Failed to load connection"}), 500)
|
|
return make_response(jsonify({"success": True, "connection": detail}), 200)
|
|
|
|
@api.doc(description="Name an account: {name}. An empty name clears it. Owner only.")
|
|
def patch(self, connection_id: str):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
name = _json_body().get("name")
|
|
if not isinstance(name, str) or len(name.strip()) > service.ACCOUNT_NAME_MAX:
|
|
return _error(f"name must be text of at most {service.ACCOUNT_NAME_MAX} characters", 400)
|
|
with db_session() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
connection = service.rename_connection(conn, row, name)
|
|
return make_response(jsonify({"success": True, "connection": connection}), 200)
|
|
|
|
@api.doc(
|
|
description=(
|
|
"Remove a connection: {sources: keep | delete, tools: delete | keep}. "
|
|
"Kept sources keep their content and stop syncing."
|
|
)
|
|
)
|
|
def delete(self, connection_id: str):
|
|
from docsgpt.api.user.sources.routes import delete_source
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
body = _json_body()
|
|
sources_mode = body.get("sources", "keep")
|
|
tools_mode = body.get("tools", "delete")
|
|
if sources_mode not in ("keep", "delete") or tools_mode not in ("keep", "delete"):
|
|
return _error("sources and tools must be keep or delete", 400)
|
|
try:
|
|
with db_session() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
to_delete = service.remove_connection(conn, row, sources=sources_mode, tools=tools_mode)
|
|
failed = [str(doc["id"]) for doc in to_delete if not delete_source(user_id, doc)]
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error removing connection: {err}", exc_info=True)
|
|
return _error("Failed to remove connection", 500)
|
|
return make_response(jsonify({"success": True, "failed_sources": failed}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/disconnect")
|
|
class ConnectionDisconnect(Resource):
|
|
@api.doc(description="Delete a connection's stored credentials; its sources and tools stay")
|
|
def post(self, connection_id: str):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
try:
|
|
with db_session() as conn:
|
|
row = ConnectorSessionsRepository(conn).get_for_user(connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
connection = service.disconnect(conn, row)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error disconnecting connection: {err}", exc_info=True)
|
|
return make_response(jsonify({"success": False, "error": "Failed to disconnect"}), 500)
|
|
return make_response(jsonify({"success": True, "connection": connection}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/setup")
|
|
class ConnectionSetup(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"Apply the connect wizard's choices: {create_tools, allow_writes?, tool_permissions?, "
|
|
"sync?: {items, frequency, name?, config?}}. allow_writes points GitHub's tool at its write "
|
|
"endpoint. config is the synced source's retrieval settings, validated like an upload's. Honours an Idempotency-Key header for the sync."
|
|
)
|
|
)
|
|
def post(self, connection_id: str):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
body = _json_body()
|
|
create_tools = body.get("create_tools", True)
|
|
allow_writes = body.get("allow_writes", False)
|
|
if not isinstance(allow_writes, bool):
|
|
return _error("allow_writes must be true or false", 400)
|
|
try:
|
|
mcp_actions = None
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
discover = bool(row) and create_tools and service.needs_mcp_discovery(conn, row)
|
|
forbidden = bool(row) and allow_writes and not service.writes_allowed(
|
|
service.load_policies(conn), catalog.connector_key_for_row(row),
|
|
)
|
|
if row is None:
|
|
return _not_found()
|
|
if service.normalize_status(row) != service.STATUS_CONNECTED:
|
|
return _error("Reconnect before setting up", 409, code="reconnect")
|
|
if forbidden:
|
|
return _error("Changes through this connector are turned off by an admin", 403,
|
|
code="writes_forbidden")
|
|
if discover:
|
|
# GitHub's tool is its MCP server: read its actions before
|
|
# the write transaction, not while holding it open.
|
|
from docsgpt.connectors.mcp import discover_builtin_actions
|
|
|
|
try:
|
|
mcp_actions = discover_builtin_actions(user_id, row, writes=allow_writes)
|
|
except service.ConnectionUnavailable:
|
|
return _error("Reconnect before setting up", 409, code="reconnect")
|
|
except Exception as err:
|
|
current_app.logger.warning(f"Could not list the MCP server's tools: {err}")
|
|
return _error("The service's tools could not be reached. Try again.", 502,
|
|
code="tools_unavailable")
|
|
with db_session() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
tools = []
|
|
if create_tools:
|
|
tools = service.ensure_connection_tools(
|
|
conn, user_id, row, permissions=body.get("tool_permissions") or None,
|
|
mcp_actions=mcp_actions, mcp_writes=allow_writes,
|
|
)
|
|
account_parameters = service.connection_parameters(row)
|
|
tool_payload = [service.serialize_tool(tool, account_parameters) for tool in tools]
|
|
sources = []
|
|
if body.get("sync"):
|
|
started = _start_sync(user_id, row, body["sync"])
|
|
if isinstance(started, tuple):
|
|
return _error(*started)
|
|
sources.append(started)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error setting up connection: {err}", exc_info=True)
|
|
return _error("Failed to set up connection", 500)
|
|
return make_response(jsonify({"success": True, "tools": tool_payload, "sources": sources}), 200)
|
|
|
|
|
|
def _start_sync(user_id: str, row: dict, sync: dict):
|
|
"""Queue the first ingest of a source synced from ``row``.
|
|
|
|
Args:
|
|
user_id: The connection's owner.
|
|
row: The connection row.
|
|
sync: ``{items, frequency, name?, config?}``. ``config`` is the source's
|
|
retrieval settings (a ``SourceConfig``), applied as an upload's are.
|
|
|
|
Returns:
|
|
The source summary, or ``(message, status)`` on a bad request.
|
|
"""
|
|
from docsgpt.api.user.sources.upload import (
|
|
_claim_task_or_get_cached,
|
|
_derive_source_id,
|
|
_parse_source_config,
|
|
_read_idempotency_key,
|
|
_scoped_idempotency_key,
|
|
)
|
|
from docsgpt.api.user.tasks import ingest_connector_task, ingest_remote
|
|
|
|
definition = catalog.get_definition(catalog.connector_key_for_row(row))
|
|
if definition is None or not definition.sync_ingestor:
|
|
return ("This connector does not sync content", 400)
|
|
items = sync.get("items") or {}
|
|
if not isinstance(items, dict):
|
|
return ("items must be an object", 400)
|
|
frequency = sync.get("frequency") or definition.default_sync_frequency
|
|
if frequency not in _FREQUENCIES:
|
|
return ("Unknown sync frequency", 400)
|
|
name = (sync.get("name") or "").strip()
|
|
source_config, config_error = _parse_source_config(sync.get("config"))
|
|
if config_error is not None:
|
|
return ("Invalid source config", 400)
|
|
if definition.sync_ingestor == "github":
|
|
from docsgpt.parser.remote.github_loader import GitHubLoader
|
|
|
|
repo = GitHubLoader.normalize_repo(str(items.get("repo_url") or ""))
|
|
if not repo:
|
|
return ("Pick a GitHub repository", 400)
|
|
items = {**items, "repo_url": repo}
|
|
name = name or repo
|
|
elif definition.sync_ingestor == "linear":
|
|
from docsgpt.connectors import linear
|
|
|
|
try:
|
|
items = linear.normalize_selection(items)
|
|
except ValueError as err:
|
|
return (str(err), 400)
|
|
name = name or linear.selection_name(items)
|
|
name = name or definition.name
|
|
# Validate before claiming the idempotency key: a rejected request must
|
|
# leave the key free for the corrected retry.
|
|
if definition.auth_kind == "oauth":
|
|
file_ids = [str(i) for i in items.get("file_ids") or [] if i]
|
|
folder_ids = [str(i) for i in items.get("folder_ids") or [] if i]
|
|
if not file_ids and not folder_ids:
|
|
return ("Pick at least one file or folder", 400)
|
|
task_fn = ingest_connector_task
|
|
kwargs = {
|
|
"job_name": name,
|
|
"user": user_id,
|
|
"source_type": definition.sync_ingestor,
|
|
"connection_id": str(row["id"]),
|
|
"file_ids": file_ids,
|
|
"folder_ids": folder_ids,
|
|
"recursive": bool(items.get("recursive", True)),
|
|
"sync_frequency": frequency,
|
|
"config": source_config,
|
|
}
|
|
else:
|
|
if definition.sync_ingestor == "linear":
|
|
# The teams and projects picked, read with the connection's MCP sign-in.
|
|
source_data = items
|
|
else:
|
|
fields = {f.key for f in definition.setup_fields}
|
|
source_data = {k: v for k, v in items.items() if k in fields and v not in (None, "")}
|
|
missing = [f.label for f in definition.setup_fields if f.required and f.key not in source_data]
|
|
if missing:
|
|
return (f"Missing: {', '.join(missing)}", 400)
|
|
task_fn = ingest_remote
|
|
kwargs = {
|
|
"source_data": source_data,
|
|
"job_name": name,
|
|
"user": user_id,
|
|
"loader": definition.sync_ingestor,
|
|
"connection_id": str(row["id"]),
|
|
"sync_frequency": frequency,
|
|
"config": source_config,
|
|
}
|
|
idempotency_key, _ = _read_idempotency_key()
|
|
scoped_key = _scoped_idempotency_key(idempotency_key, user_id)
|
|
task_id = None
|
|
if scoped_key:
|
|
task_id, cached = _claim_task_or_get_cached(scoped_key, "connection_setup_sync")
|
|
if cached is not None:
|
|
return {"id": cached.get("source_id"), "task_id": cached.get("task_id"), "name": name}
|
|
source_id = str(_derive_source_id(scoped_key)) if scoped_key else str(uuid.uuid4())
|
|
options = {"task_id": task_id} if task_id else {}
|
|
task = task_fn.apply_async(
|
|
kwargs={**kwargs, "idempotency_key": scoped_key, "source_id": source_id}, **options,
|
|
)
|
|
return {"id": source_id, "task_id": task_id or task.id, "name": name, "sync_frequency": frequency}
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/repositories")
|
|
class ConnectionRepositories(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"GitHub: the repositories the connection can read, for the sync picker. "
|
|
"install_url is where a GitHub App sign-in chooses more repositories."
|
|
)
|
|
)
|
|
def get(self, connection_id: str):
|
|
from docsgpt.connectors import github
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None or catalog.connector_key_for_row(row) != "github":
|
|
return _not_found()
|
|
app_sign_in = (row.get("auth_kind") or "") == "oauth"
|
|
try:
|
|
token = service.access_credentials(row).get("access_token")
|
|
repositories = github.list_repositories(token or "", app=app_sign_in)
|
|
except service.ConnectionUnavailable:
|
|
return _error("Reconnect to continue", 409, code="reconnect")
|
|
except github.TokenRejected as err:
|
|
service.mark_reconnect_needed(connection_id, str(err))
|
|
return _error("Reconnect to continue", 409, code="reconnect")
|
|
except service.TransientConnectionError:
|
|
return _error("GitHub is not responding. Try again.", 503)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error listing GitHub repositories: {err}", exc_info=True)
|
|
return _error("Failed to list repositories", 502)
|
|
install_url = None
|
|
if app_sign_in:
|
|
from docsgpt.core.settings import settings
|
|
|
|
slug = settings.GITHUB_APP_SLUG
|
|
install_url = f"https://github.com/apps/{slug}/installations/new" if slug else None
|
|
return make_response(
|
|
jsonify({"success": True, "repositories": repositories, "install_url": install_url}), 200,
|
|
)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/linear")
|
|
class LinearWorkspace(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"Linear: the teams and projects the connection can see, for the sync picker. "
|
|
"Read through Linear's MCP server with the connection's sign-in."
|
|
)
|
|
)
|
|
def get(self, connection_id: str):
|
|
from docsgpt.connectors import linear, mcp
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None or catalog.connector_key_for_row(row) != linear.LINEAR_CONNECTOR:
|
|
return _not_found()
|
|
try:
|
|
workspace = mcp.run_connection_session(row, linear.mcp_url(), linear.list_workspace)
|
|
except service.ConnectionUnavailable:
|
|
return _error("Reconnect to continue", 409, code="reconnect")
|
|
except service.TransientConnectionError:
|
|
return _error("Linear is not responding. Try again.", 503)
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error listing Linear teams: {err}", exc_info=True)
|
|
return _error("Failed to list Linear teams", 502)
|
|
return make_response(jsonify({"success": True, **workspace}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/reconnect")
|
|
class ConnectionReconnect(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"OAuth: returns an authorization URL for the same account. "
|
|
"API key: accepts {credentials} and replaces the stored ones."
|
|
)
|
|
)
|
|
def post(self, connection_id: str):
|
|
from docsgpt.api.connector.routes import build_authorization
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
body = _json_body()
|
|
try:
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
key = catalog.connector_key_for_row(row)
|
|
definition = catalog.get_definition(key)
|
|
auth_kind = row.get("auth_kind") or (definition.auth_kind if definition else None)
|
|
if auth_kind == "oauth":
|
|
started = build_authorization(row["provider"], user_id, connection_id)
|
|
return make_response(jsonify({"success": True, "kind": "oauth", **started}), 200)
|
|
if auth_kind == "mcp_oauth":
|
|
# The MCP client runs the OAuth dance (dynamic registration,
|
|
# PKCE); the frontend starts it through /api/mcp_server/test.
|
|
return make_response(
|
|
jsonify({"success": True, "kind": "mcp_oauth", "server_url": row.get("server_url")}), 200,
|
|
)
|
|
credentials = body.get("credentials")
|
|
if not isinstance(credentials, dict) or not credentials:
|
|
return _error("credentials are required", 400)
|
|
service.ensure_can_store_credentials()
|
|
with db_session() as conn:
|
|
locked = ConnectorSessionsRepository(conn).get_for_update(connection_id)
|
|
# read_secrets, not load_secrets: flagging an unreadable row
|
|
# would write it from a second transaction while this one
|
|
# holds its lock. The new credentials replace it anyway.
|
|
try:
|
|
stored = service.read_secrets(locked)
|
|
except CredentialDecryptionError:
|
|
stored = {}
|
|
merged = {**(stored.get("credentials") or {}), **{k: v for k, v in credentials.items() if v}}
|
|
service.write_secrets(
|
|
conn, locked, {**stored, "credentials": merged},
|
|
status=service.STATUS_CONNECTED, last_error=None,
|
|
)
|
|
service.resume_sources(conn, connection_id)
|
|
connection = service.serialize_connection(ConnectorSessionsRepository(conn).get(connection_id))
|
|
except service.EncryptionKeyNotConfigured as err:
|
|
return _error(str(err), 400, code="encryption_key_default")
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error reconnecting: {err}", exc_info=True)
|
|
return _error("Failed to reconnect", 500)
|
|
return make_response(jsonify({"success": True, "kind": "api_key", "connection": connection}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/picker-token")
|
|
class ConnectionPickerToken(Resource):
|
|
@api.doc(description="A short-lived access token for a browser-side file picker. Owner only; never a refresh token.")
|
|
def post(self, connection_id: str):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None or (row.get("auth_kind") or "oauth") != "oauth":
|
|
return _not_found()
|
|
try:
|
|
token = service.picker_token(connection_id)
|
|
except service.ConnectionUnavailable:
|
|
return _error("Reconnect to continue", 409, code="reconnect")
|
|
except service.TransientConnectionError:
|
|
return _error("The provider is not responding. Try again.", 503)
|
|
return make_response(jsonify({"success": True, **token}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/claim")
|
|
class ConnectionClaim(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"One-time link of a legacy browser session token ({provider, session_token}) "
|
|
"to the caller's connection. Removed next release."
|
|
)
|
|
)
|
|
def post(self):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
body = _json_body()
|
|
provider, token = body.get("provider"), body.get("session_token")
|
|
if not provider or not token:
|
|
return _error("provider and session_token are required", 400)
|
|
with db_readonly() as conn:
|
|
row = service.claim_session_token(conn, user_id, str(provider), str(token))
|
|
if row is None:
|
|
return _not_found()
|
|
return make_response(jsonify({"success": True, "connection_id": str(row["id"])}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/tools/<string:tool_id>/permissions")
|
|
class ConnectionToolPermissions(Resource):
|
|
@api.doc(description="Set per-action permissions: {permissions: {action: always | ask | off}}")
|
|
def put(self, connection_id: str, tool_id: str):
|
|
from docsgpt.connectors.permissions import PERMISSIONS
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
permissions = _json_body().get("permissions")
|
|
if not isinstance(permissions, dict) or any(p not in PERMISSIONS for p in permissions.values()):
|
|
return _error("permissions must map action names to always, ask or off", 400)
|
|
with db_session() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
tool = service.set_tool_permissions(conn, user_id, connection_id, tool_id, permissions)
|
|
if tool is None:
|
|
return _not_found()
|
|
payload = service.serialize_tool(tool, service.connection_parameters(row))
|
|
return make_response(jsonify({"success": True, "tool": payload}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/tools/<string:tool_id>/parameters")
|
|
class ConnectionToolParameters(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"Fix or release an action's parameters: {action, parameters: {name: value | null}}. "
|
|
"A value is sent on every call and hidden from the model; null lets the model decide. Owner only."
|
|
)
|
|
)
|
|
def put(self, connection_id: str, tool_id: str):
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
body = _json_body()
|
|
action, pins = body.get("action"), body.get("parameters")
|
|
if not isinstance(action, str) or not isinstance(pins, dict) or not pins:
|
|
return _error("Send the action and a map of parameters to values or null", 400)
|
|
with db_session() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
try:
|
|
tool = service.set_tool_parameters(conn, user_id, connection_id, tool_id, action, pins)
|
|
except ValueError as err:
|
|
return _error(str(err), 400)
|
|
if tool is None:
|
|
return _not_found()
|
|
payload = service.serialize_tool(tool, service.connection_parameters(row))
|
|
return make_response(jsonify({"success": True, "tool": payload}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/refresh-tools")
|
|
class ConnectionRefreshTools(Resource):
|
|
@api.doc(description="MCP: re-scan the server's actions and return what was added and removed")
|
|
def post(self, connection_id: str):
|
|
from docsgpt.connectors.mcp import refresh_mcp_tools
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
try:
|
|
diff = refresh_mcp_tools(user_id, row)
|
|
except service.ConnectionUnavailable:
|
|
return _error("Reconnect to continue", 409, code="reconnect")
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error refreshing MCP tools: {err}", exc_info=True)
|
|
return _error("Failed to refresh tools", 502)
|
|
return make_response(jsonify({"success": True, **diff}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/<string:connection_id>/writes")
|
|
class ConnectionWrites(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"GitHub: let agents make changes through this connection, or only read: {allow}. "
|
|
"Re-reads the tool's actions from the matching endpoint. Owner only."
|
|
)
|
|
)
|
|
def put(self, connection_id: str):
|
|
from docsgpt.connectors.mcp import NoMcpTool, set_builtin_writes
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
allow = _json_body().get("allow")
|
|
if not isinstance(allow, bool):
|
|
return _error("allow must be true or false", 400)
|
|
with db_readonly() as conn:
|
|
row = _owned(conn, connection_id, user_id)
|
|
if row is None:
|
|
return _not_found()
|
|
try:
|
|
result = set_builtin_writes(user_id, row, allow)
|
|
except service.WritesForbidden as err:
|
|
return _error(str(err), 403, code="writes_forbidden")
|
|
except NoMcpTool as err:
|
|
return _error(str(err), 409, code="no_tools")
|
|
except ValueError as err:
|
|
return _error(str(err), 400)
|
|
except service.ConnectionUnavailable:
|
|
return _error("Reconnect to continue", 409, code="reconnect")
|
|
except Exception as err:
|
|
current_app.logger.error(f"Error switching write access: {err}", exc_info=True)
|
|
return _error("The service's tools could not be reached. Try again.", 502, code="tools_unavailable")
|
|
return make_response(jsonify({"success": True, **result}), 200)
|
|
|
|
|
|
@connections_ns.route("/connections/tools/<string:tool_id>/credential-mode")
|
|
class ToolCredentialMode(Resource):
|
|
@api.doc(
|
|
description=(
|
|
"Whose account a shared connection-backed tool uses: {mode: owner | member}. "
|
|
"Owner only; refused when an admin forces a mode for the connector."
|
|
)
|
|
)
|
|
def put(self, tool_id: str):
|
|
from docsgpt.connectors.resolve import MODE_MEMBER, MODE_OWNER
|
|
from docsgpt.storage.db.repositories.user_tools import UserToolsRepository
|
|
|
|
user_id = _user_id()
|
|
if not user_id:
|
|
return _unauthorized()
|
|
mode = _json_body().get("mode")
|
|
if mode not in (MODE_OWNER, MODE_MEMBER):
|
|
return _error("mode must be owner or member", 400)
|
|
with db_session() as conn:
|
|
tools = UserToolsRepository(conn)
|
|
tool = tools.get_any(tool_id, user_id)
|
|
if tool is None or tool.get("user_id") != user_id or not tool.get("connection_id"):
|
|
return _error("Tool not found", 404)
|
|
connection = ConnectorSessionsRepository(conn).get(str(tool["connection_id"]))
|
|
forced = service.forced_credential_mode(
|
|
conn, catalog.connector_key_for_row(connection) if connection else None,
|
|
)
|
|
if forced and forced != mode:
|
|
return _error("An admin sets this for every share", 409, code="forced", mode=forced)
|
|
tools.update(str(tool["id"]), user_id, {"credential_mode": mode})
|
|
return make_response(jsonify({"success": True, "mode": mode}), 200)
|