airbyte_ops_mcp.mcp.devin_secret_request

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for on-demand Devin secret request and discovery.
  3
  4Tools:
  5
  6- **list_devin_secrets**: Lists all available secret names in the
  7  1Password vault, so the agent can discover valid secret aliases.
  8- **request_devin_secret**: Two-phase approval workflow to request
  9  and deliver a secret.
 10
 11## MCP reference
 12
 13.. include:: ../../../docs/mcp-generated/devin_secret_request.md
 14    :start-line: 2
 15"""
 16
 17# NOTE: We intentionally do NOT use `from __future__ import annotations` here.
 18# FastMCP has issues resolving forward references when PEP 563 deferred annotations
 19# are used. See: https://github.com/jlowin/fastmcp/issues/905
 20
 21__all__: list[str] = []
 22
 23import json
 24import logging
 25import re
 26from typing import Annotated
 27
 28import requests
 29from fastmcp import FastMCP
 30from fastmcp_extensions import mcp_tool, register_mcp_tools
 31from pydantic import BaseModel, Field
 32
 33from airbyte_ops_mcp.github_actions import (
 34    download_job_logs,
 35    get_workflow_jobs,
 36    resolve_default_workflow_branch,
 37    trigger_workflow_dispatch,
 38    wait_for_workflow_completion,
 39)
 40from airbyte_ops_mcp.github_api import resolve_ci_trigger_github_token
 41
 42logger = logging.getLogger(__name__)
 43
 44# ---------------------------------------------------------------------------
 45# Constants
 46# ---------------------------------------------------------------------------
 47
 48WORKFLOW_REPO_OWNER = "airbytehq"
 49WORKFLOW_REPO_NAME = "airbyte-ops-mcp"
 50WORKFLOW_FILE = "devin-secret-request.yml"
 51WORKFLOW_DEFAULT_BRANCH = "main"
 52
 53
 54_SESSION_ID_PATTERN = re.compile(r"[0-9a-fA-F]{32}")
 55
 56
 57class SecretListResponse(BaseModel):
 58    """Response from the list_devin_secrets tool."""
 59
 60    success: bool = Field(description="Whether the operation succeeded")
 61    message: str = Field(description="Human-readable status message")
 62    available_secrets: list[str] = Field(
 63        default_factory=list,
 64        description="Sorted list of available secret names in the vault",
 65    )
 66
 67
 68class SecretRequestResponse(BaseModel):
 69    """Response from the request_devin_secret tool."""
 70
 71    success: bool = Field(description="Whether the operation succeeded")
 72    phase: str = Field(
 73        description=(
 74            "Current phase: 'approval_requested' (Phase 1) or "
 75            "'delivery_dispatched' (Phase 2)"
 76        ),
 77    )
 78    message: str = Field(description="Human-readable status message")
 79    secret_alias: str = Field(description="The requested secret alias")
 80    session_id: str = Field(description="The Devin session ID")
 81    workflow_url: str | None = Field(
 82        default=None,
 83        description="URL to the GitHub Actions workflow",
 84    )
 85    run_id: int | None = Field(
 86        default=None,
 87        description="GitHub Actions workflow run ID",
 88    )
 89    run_url: str | None = Field(
 90        default=None,
 91        description="Direct URL to the GitHub Actions workflow run",
 92    )
 93    request_id: str | None = Field(
 94        default=None,
 95        description=(
 96            "Unique request identifier (UUID). Returned in Phase 1; "
 97            "pass it back in Phase 2 for replay-protection validation."
 98        ),
 99    )
100
101
102@mcp_tool(
103    read_only=False,
104    idempotent=False,
105    open_world=True,
106)
107def list_devin_secrets() -> SecretListResponse:
108    """List all available secret names in the 1Password vault.
109
110    Returns the sorted list of item titles from the
111    'devin-on-demand-secrets' vault. Use this to discover valid
112    secret aliases before calling request_devin_secret.
113
114    This dispatches a GitHub Actions workflow (which has the
115    1Password credentials), waits for it to complete, then reads
116    the list from the job logs.
117    """
118    return _list_secrets_via_workflow()
119
120
121@mcp_tool(
122    read_only=False,
123    idempotent=False,
124    open_world=True,
125)
126def request_devin_secret(
127    secret_alias: Annotated[
128        str,
129        "The name of the secret to request. This must exactly match an item "
130        "title in the 'devin-on-demand-secrets' 1Password vault.",
131    ],
132    session_url: Annotated[
133        str,
134        "Your Devin session URL (e.g. 'https://app.devin.ai/sessions/abc123...'). "
135        "Use the session URL from your system prompt.",
136    ],
137    approval_evidence_url: Annotated[
138        str | None,
139        "Slack approval record URL "
140        "(https://<workspace>.slack.com/archives/...). "
141        "Leave empty for Phase 1 (requesting approval). Provide the "
142        "Slack URL for Phase 2 (delivering the secret after approval).",
143    ] = None,
144    target_approver: Annotated[
145        str | None,
146        "Person to notify for approval (GitHub handle, email, or Slack user ID). "
147        "Required for Phase 1 (approval request).",
148    ] = None,
149    request_id: Annotated[
150        str | None,
151        "Request ID returned by Phase 1. Pass it back in Phase 2 "
152        "so the approval record can be validated against the original request. "
153        "Leave empty for Phase 1.",
154    ] = None,
155) -> SecretRequestResponse:
156    """Request a secret on demand via an approval workflow.
157
158    This tool operates in two phases:
159
160    **Phase 1** (no approval_evidence_url): Dispatches a GitHub Actions
161    workflow that validates the secret name against the 1Password vault
162    and, if valid, sends a Slack approval request. If the secret name is
163    not found, returns immediately with the list of available secret
164    names so you can correct any typos.
165
166    **Phase 2** (with approval_evidence_url): After a human approves the
167    request, call this tool again with the approval evidence URL. This
168    triggers a GitHub Actions workflow that reads the secret from
169    1Password and sends you a time-limited share link.
170    Open the link in your browser to view and copy the secret.
171
172    Typical workflow:
173    0. (Optional) Call list_devin_secrets first to see available names.
174    1. Call this tool without approval_evidence_url to request approval.
175    2. Note the `request_id` in the response.
176    3. Wait for a human to approve the request in Slack.
177    4. Obtain the approval evidence URL (Slack approval record URL).
178    5. Call this tool again with the approval_evidence_url **and** the
179       request_id from step 2.
180    6. You will receive a 1Password share link -- open it in your
181       browser to view and copy the secret values.
182    """
183    # Extract session ID from URL
184    match = _SESSION_ID_PATTERN.search(session_url)
185    if not match:
186        return SecretRequestResponse(
187            success=False,
188            phase="error",
189            message=(
190                f"No valid session ID found in URL: {session_url}. "
191                "Expected a 32-character hex string."
192            ),
193            secret_alias=secret_alias,
194            session_id="",
195        )
196    session_id = match.group(0)
197
198    if not approval_evidence_url:
199        # Phase 1: Dispatch the request workflow (validates secret name
200        # inline using op CLI, then sends Slack approval if valid).
201        if not target_approver:
202            return SecretRequestResponse(
203                success=False,
204                phase="error",
205                message=(
206                    "target_approver is required when requesting approval "
207                    "(no approval_evidence_url provided)."
208                ),
209                secret_alias=secret_alias,
210                session_id=session_id,
211            )
212
213        return _request_secret_via_workflow(
214            secret_alias=secret_alias,
215            session_id=session_id,
216            session_url=session_url,
217            target_approver=target_approver,
218        )
219
220    # Phase 2: Deliver secret via GitHub Actions workflow
221    token = resolve_ci_trigger_github_token()
222
223    workflow_inputs: dict[str, str] = {
224        "action": "deliver",
225        "secret_alias": secret_alias,
226        "session_id": session_id,
227        "approval_evidence_url": approval_evidence_url,
228    }
229    if request_id:
230        workflow_inputs["expected_request_id"] = request_id
231
232    result = trigger_workflow_dispatch(
233        owner=WORKFLOW_REPO_OWNER,
234        repo=WORKFLOW_REPO_NAME,
235        workflow_file=WORKFLOW_FILE,
236        ref=resolve_default_workflow_branch(WORKFLOW_DEFAULT_BRANCH),
237        inputs=workflow_inputs,
238        token=token,
239    )
240
241    view_url = result.run_url or result.workflow_url
242    return SecretRequestResponse(
243        success=True,
244        phase="delivery_dispatched",
245        message=(
246            f"Secret delivery workflow dispatched for '{secret_alias}'. "
247            f"The workflow will read the secret from 1Password and send "
248            f"you a time-limited share link. Once you receive the link, "
249            f"open it in your browser to view and copy the secret. "
250            f"View progress: {view_url}"
251        ),
252        secret_alias=secret_alias,
253        session_id=session_id,
254        workflow_url=result.workflow_url,
255        run_id=result.run_id,
256        run_url=result.run_url,
257        request_id=request_id,
258    )
259
260
261def _request_secret_via_workflow(
262    secret_alias: str,
263    session_id: str,
264    session_url: str,
265    target_approver: str,
266) -> SecretRequestResponse:
267    """Dispatch the request workflow, wait, and parse the result from job logs.
268
269    The workflow validates the secret alias against the vault inline,
270    then sends the Slack approval if valid.  On a bad alias the workflow
271    fails and the job logs contain a JSON object with `available_secrets`.
272    """
273    token = resolve_ci_trigger_github_token()
274
275    dispatch_result = trigger_workflow_dispatch(
276        owner=WORKFLOW_REPO_OWNER,
277        repo=WORKFLOW_REPO_NAME,
278        workflow_file=WORKFLOW_FILE,
279        ref=resolve_default_workflow_branch(WORKFLOW_DEFAULT_BRANCH),
280        inputs={
281            "action": "request",
282            "secret_alias": secret_alias,
283            "session_id": session_id,
284            "target_approver": target_approver,
285        },
286        token=token,
287    )
288    if not dispatch_result.run_id:
289        return SecretRequestResponse(
290            success=False,
291            phase="error",
292            message=(
293                "Workflow dispatched but no run ID returned. "
294                f"Check: {dispatch_result.workflow_url}"
295            ),
296            secret_alias=secret_alias,
297            session_id=session_id,
298            workflow_url=dispatch_result.workflow_url,
299        )
300
301    run_status = wait_for_workflow_completion(
302        owner=WORKFLOW_REPO_OWNER,
303        repo=WORKFLOW_REPO_NAME,
304        run_id=dispatch_result.run_id,
305        token=token,
306    )
307
308    # Download logs from the validation job (multi-job workflow)
309    raw_logs = _download_run_logs(
310        dispatch_result.run_id, token, job_name="Validate Secret Name"
311    )
312
313    if run_status.succeeded:
314        # Parse the approval-requested JSON from the logs
315        result_data = _find_json_in_logs(raw_logs, "phase") if raw_logs else None
316        request_id = result_data.get("request_id") if result_data else None
317        view_url = run_status.run_url or dispatch_result.workflow_url
318        return SecretRequestResponse(
319            success=True,
320            phase="approval_requested",
321            message=(
322                f"Approval request for secret '{secret_alias}' sent to "
323                f"#human-in-the-loop. Waiting for human approval. "
324                f"Once approved, call this tool again with the "
325                f"approval_evidence_url to deliver the secret. "
326                f"View progress: {view_url}"
327            ),
328            secret_alias=secret_alias,
329            session_id=session_id,
330            workflow_url=dispatch_result.workflow_url,
331            run_id=dispatch_result.run_id,
332            run_url=run_status.run_url,
333            request_id=request_id,
334        )
335
336    # Workflow failed — check if it was a validation failure
337    error_data = _find_json_in_logs(raw_logs, "available_secrets") if raw_logs else None
338    if error_data:
339        available = error_data.get("available_secrets", [])
340        formatted = ", ".join(f"`{s}`" for s in available)
341        return SecretRequestResponse(
342            success=False,
343            phase="validation_failed",
344            message=(
345                f"Secret '{secret_alias}' not found in the vault. "
346                f"Available secrets: {formatted}"
347            ),
348            secret_alias=secret_alias,
349            session_id=session_id,
350            workflow_url=dispatch_result.workflow_url,
351            run_id=dispatch_result.run_id,
352            run_url=run_status.run_url,
353        )
354
355    # Generic workflow failure
356    return SecretRequestResponse(
357        success=False,
358        phase="error",
359        message=(
360            f"Request workflow failed (conclusion={run_status.conclusion}). "
361            f"See: {run_status.run_url}"
362        ),
363        secret_alias=secret_alias,
364        session_id=session_id,
365        workflow_url=dispatch_result.workflow_url,
366        run_id=dispatch_result.run_id,
367        run_url=run_status.run_url,
368    )
369
370
371def _download_run_logs(
372    run_id: int,
373    token: str,
374    *,
375    job_name: str | None = None,
376) -> str | None:
377    """Best-effort download of a job's logs for a workflow run.
378
379    Args:
380        run_id: GitHub Actions workflow run ID.
381        token: GitHub API token used for log download. Note: job listing
382            uses `get_workflow_jobs` which resolves its own token via
383            `resolve_ci_trigger_github_token()`.
384        job_name: If provided, find the job whose name contains this
385            substring (case-insensitive). Skipped jobs are always
386            excluded. Falls back to the first non-skipped job.
387    """
388    try:
389        jobs = get_workflow_jobs(
390            owner=WORKFLOW_REPO_OWNER,
391            repo=WORKFLOW_REPO_NAME,
392            run_id=run_id,
393        )
394        # Filter out skipped jobs (common in multi-job conditional workflows)
395        active_jobs = [j for j in jobs if j.conclusion != "skipped"]
396        if not active_jobs:
397            return None
398
399        target = active_jobs[0]  # default: first non-skipped job
400        if job_name:
401            needle = job_name.lower()
402            for j in active_jobs:
403                if needle in j.name.lower():
404                    target = j
405                    break
406
407        return download_job_logs(
408            owner=WORKFLOW_REPO_OWNER,
409            repo=WORKFLOW_REPO_NAME,
410            job_id=target.job_id,
411            token=token,
412        )
413    except (requests.HTTPError, ValueError) as exc:
414        logger.warning("Failed to download job logs for run %s: %s", run_id, exc)
415        return None
416
417
418def _list_secrets_via_workflow() -> SecretListResponse:
419    """Dispatch the list workflow, wait for completion, and parse titles from job logs."""
420    token = resolve_ci_trigger_github_token()
421
422    # 1. Dispatch the workflow with action="list"
423    dispatch_result = trigger_workflow_dispatch(
424        owner=WORKFLOW_REPO_OWNER,
425        repo=WORKFLOW_REPO_NAME,
426        workflow_file=WORKFLOW_FILE,
427        ref=resolve_default_workflow_branch(WORKFLOW_DEFAULT_BRANCH),
428        inputs={"action": "list", "session_id": "0" * 32},
429        token=token,
430    )
431    if not dispatch_result.run_id:
432        return SecretListResponse(
433            success=False,
434            message=(
435                "Workflow dispatched but no run ID returned. "
436                f"Check: {dispatch_result.workflow_url}"
437            ),
438        )
439
440    # 2. Wait for the workflow to complete
441    run_status = wait_for_workflow_completion(
442        owner=WORKFLOW_REPO_OWNER,
443        repo=WORKFLOW_REPO_NAME,
444        run_id=dispatch_result.run_id,
445        token=token,
446    )
447    if not run_status.succeeded:
448        return SecretListResponse(
449            success=False,
450            message=(
451                f"Workflow run failed (conclusion={run_status.conclusion}). "
452                f"See: {run_status.run_url}"
453            ),
454        )
455
456    # 3. Find the job and download its logs
457    jobs = get_workflow_jobs(
458        owner=WORKFLOW_REPO_OWNER,
459        repo=WORKFLOW_REPO_NAME,
460        run_id=dispatch_result.run_id,
461    )
462    if not jobs:
463        return SecretListResponse(
464            success=False,
465            message="Workflow completed but no jobs found.",
466        )
467
468    # Find the list job (multi-job workflow; skip skipped jobs)
469    active_jobs = [j for j in jobs if j.conclusion != "skipped"]
470    if not active_jobs:
471        return SecretListResponse(
472            success=False,
473            message="Workflow completed but all jobs were skipped.",
474        )
475
476    target_job = active_jobs[0]
477    for j in active_jobs:
478        if "list" in j.name.lower():
479            target_job = j
480            break
481
482    raw_logs = download_job_logs(
483        owner=WORKFLOW_REPO_OWNER,
484        repo=WORKFLOW_REPO_NAME,
485        job_id=target_job.job_id,
486        token=token,
487    )
488
489    # 4. Parse JSON output from the logs
490    data = _find_json_in_logs(raw_logs, "available_secrets")
491    if data is None:
492        return SecretListResponse(
493            success=False,
494            message=(
495                "Could not parse secret list from workflow logs. "
496                f"See: {run_status.run_url}"
497            ),
498        )
499
500    secrets = data.get("available_secrets", [])
501    titles = [str(t) for t in secrets] if isinstance(secrets, list) else []
502    return SecretListResponse(
503        success=True,
504        message=f"Found {len(titles)} available secrets in the vault.",
505        available_secrets=sorted(titles),
506    )
507
508
509def _find_json_in_logs(raw_logs: str, required_key: str) -> dict | None:
510    """Find the first JSON object in job logs that contains *required_key*.
511
512    GitHub Actions job logs prefix each line with a timestamp.  We scan
513    every line looking for a JSON object that contains the given key.
514    Returns the parsed dict, or `None` if not found.
515    """
516    for line in raw_logs.splitlines():
517        stripped = line.strip()
518        if not stripped.startswith("{"):
519            idx = stripped.find("{")
520            if idx < 0:
521                continue
522            stripped = stripped[idx:]
523        try:
524            data = json.loads(stripped)
525        except json.JSONDecodeError:
526            continue
527        if isinstance(data, dict) and required_key in data:
528            return data
529    return None
530
531
532def register_devin_secret_request_tools(app: FastMCP) -> None:
533    """Register Devin secret request tools with the FastMCP app."""
534    register_mcp_tools(app, mcp_module=__name__)