open-swe/agent/dashboard/admin.py
Johannes du Plessis f539962c73
feat: server-side Datadog/LangSmith observability tools + team creds [closes OPE-54] (#1476)
* feat: server-side Datadog/LangSmith observability tools + team creds

Add team-wide observability credential settings (Datadog DD_SITE/API/APP
keys, LangSmith API key) stored encrypted server-side, with an admin
dashboard section to connect/disconnect each provider. When connected,
get_agent loads read-only observability tools server-side: Datadog via its
hosted MCP server (langchain-mcp-adapters, toolsets=core) and LangSmith
read tools (langsmith_get_trace, langsmith_list_runs). Credentials live in
the LangGraph server process and are never exposed to the sandbox.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

* fix: address review on observability tools

Address PR review feedback:
- Authorize observability tools per triggering user (admins + the
  OBSERVABILITY_AUTHORIZED_EMAILS allowlist) so prompt-injected runs from
  untrusted contributors can't reach team Datadog/LangSmith data.
- Use the documented Datadog MCP auth headers DD_API_KEY / DD_APPLICATION_KEY.
- Store each provider's credentials under its own store key to avoid a
  read-modify-write race dropping the other provider on concurrent saves.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

* fix: async email resolution in observability authorization gate

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
2026-06-10 11:07:42 -07:00

33 lines
1,017 B
Python

"""Admin email gate driven by the CONFIGURED_ADMINS env var."""
from __future__ import annotations
import os
def _admin_emails() -> frozenset[str]:
raw = os.environ.get("CONFIGURED_ADMINS", "")
return frozenset(e.strip().lower() for e in raw.split(",") if e.strip())
def is_admin(email: str | None) -> bool:
if not email:
return False
return email.strip().lower() in _admin_emails()
def _observability_emails() -> frozenset[str]:
raw = os.environ.get("OBSERVABILITY_AUTHORIZED_EMAILS", "")
return frozenset(e.strip().lower() for e in raw.split(",") if e.strip())
def is_observability_authorized(email: str | None) -> bool:
"""Whether ``email`` may use the team observability tools.
Admins always qualify; additional non-admin emails can be allow-listed via
``OBSERVABILITY_AUTHORIZED_EMAILS``.
"""
if not email:
return False
normalized = email.strip().lower()
return normalized in _admin_emails() or normalized in _observability_emails()