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