scan_server_card

Instructions

Implementation Reference

• bawbel_mcp/server.py:233-285 (handler)

  The handler function for the scan_server_card tool. It fetches .well-known/mcp.json from the given server URL, writes it to a temp file, runs the bawbel CLI scanner on it, formats the results, and optionally suggests a conformance check.

  @mcp.tool()
  async def scan_server_card(url: str) -> str:
      """
      Fetch and scan an MCP server-card for security vulnerabilities.
  
      Fetches .well-known/mcp.json from the given server URL and scans
      all tool descriptions, parameter descriptions, and config schemas
      for AVE vulnerabilities before your agent connects.
  
      This is the primary tool to run before adding any MCP server to
      your configuration. A poisoned server-card injects behavioral
      instructions at the discovery layer, before any tool call is made.
  
      Args:
          url: Base URL of the MCP server (e.g. https://api.example.com)
      """
      if not url.startswith(("http://", "https://")):
          return "Error: URL must start with http:// or https://"
  
      # Try server-card path first, then fall back to direct URL
      server_card_url = url.rstrip("/") + "/.well-known/mcp.json"
      content, err = _fetch_url(server_card_url)
  
      if not content:
          # Try direct URL (might be a direct link to mcp.json)
          content, err = _fetch_url(url)
          if not content:
              return f"Error: Could not fetch server-card from {server_card_url}\n{err}"
  
      with tempfile.NamedTemporaryFile(
          mode="w", suffix=".json", prefix="bawbel_mcp_ssc_",
          delete=False, encoding="utf-8"
      ) as f:
          f.write(content)
          tmp_path = f.name
  
      try:
          result = _run_bawbel(["scan"], input_file=tmp_path)
      finally:
          Path(tmp_path).unlink(missing_ok=True)
  
      output = [f"Server-card scan: {server_card_url}", ""]
      output.append(_format_scan_result(result))
  
      # Suggest conformance check
      if not result.get("error"):
          output.append("")
          output.append(
              f"Tip: run check_conformance('{url}') to score "
              "this server against the MCP spec."
          )
  
      return "\n".join(output)

• bawbel_mcp/server.py:233-234 (registration)

  The tool is registered via the @mcp.tool() decorator from FastMCP on the scan_server_card async function.

  @mcp.tool()
  async def scan_server_card(url: str) -> str:

• bawbel_mcp/server.py:54-103 (helper)

  Helper function that runs the bawbel CLI with arguments and input file, parsing JSON output. Used by scan_server_card to execute the scan command.

  def _run_bawbel(args: list[str], input_file: Optional[str] = None) -> dict:
      """
      Run bawbel CLI and return parsed JSON output.
      Returns error dict on failure.
      """
      cmd = ["bawbel"] + args + ["--format", "json"]
      if input_file:
          cmd.append(input_file)
  
      try:
          result = subprocess.run(  # nosec B603  # noqa: S603
              cmd,
              capture_output=True,
              text=True,
              timeout=60,
          )
          raw = result.stdout.strip()
          if not raw:
              return {
                  "error": result.stderr.strip() or "Scanner produced no output",
                  "findings": [],
                  "toxic_flows": [],
                  "risk_score": 0,
              }
  
          # JSON output is a list of file results
          start = raw.find("[")
          if start < 0:
              return {"error": raw[:300], "findings": [], "toxic_flows": [], "risk_score": 0}
  
          results = json.loads(raw[start:])
          if results:
              return results[0]
          return {"findings": [], "toxic_flows": [], "risk_score": 0}
  
      except subprocess.TimeoutExpired:
          return {"error": "Scan timeout (60s)", "findings": [], "toxic_flows": [], "risk_score": 0}
      except json.JSONDecodeError as e:
          return {"error": f"Parse error: {e}", "findings": [], "toxic_flows": [], "risk_score": 0}
      except FileNotFoundError:
          return {
              "error": (
                  "bawbel CLI not found. "
                  "Install with: pip install bawbel-scanner"
              ),
              "findings": [],
              "toxic_flows": [],
              "risk_score": 0,
          }

• bawbel_mcp/server.py:105-120 (helper)

  Helper function that fetches URL content. Used by scan_server_card to download the server card (mcp.json) from the target server.

  def _fetch_url(url: str) -> tuple[Optional[str], Optional[str]]:
      """Fetch content from a URL. Returns (content, error)."""
      try:
          req = urllib.request.Request(
              url,
              headers={"User-Agent": "bawbel-mcp/1.0 (https://bawbel.io)"},
          )
          with urllib.request.urlopen(req, timeout=10) as r:  # nosec B310  # noqa: S310
              return r.read(MAX_CONTENT_BYTES).decode("utf-8", errors="replace"), None
      except urllib.error.HTTPError as e:
          return None, f"HTTP {e.code}: {e.reason}"
      except urllib.error.URLError as e:
          return None, f"URL error: {e.reason}"
      except Exception as e:  # noqa: BLE001
          return None, str(e)

• bawbel_mcp/server.py:134-187 (helper)

  Helper function that formats scan results into a human-readable string. Called by scan_server_card to produce the final output.

  def _format_scan_result(result: dict) -> str:
      """Format scan result for human-readable MCP response."""
      if result.get("error"):
          return f"Error: {result['error']}"
  
      findings = result.get("findings", [])
      toxic_flows = result.get("toxic_flows", [])
      risk_score = result.get("risk_score", 0)
      max_severity = result.get("max_severity", "NONE")
  
      lines = []
  
      if not findings and not toxic_flows:
          lines.append("Clean: no findings detected.")
          lines.append(f"Risk score: {risk_score:.1f}/10")
          return "\n".join(lines)
  
      lines.append(f"Risk score: {risk_score:.1f}/10  ({max_severity})")
      lines.append(f"Findings: {len(findings)}  Toxic flows: {len(toxic_flows)}")
      lines.append("")
  
      if findings:
          lines.append("FINDINGS")
          lines.append("-" * 50)
          for f in findings:
              sev = f.get("severity", "?")
              ave_id = f.get("ave_id", "")
              title = f.get("title", "")
              line_no = f.get("line_number")
              owasp_mcp = ", ".join(f.get("owasp_mcp", []))
              lines.append(f"[{sev}] {ave_id}  {title}")
              if line_no:
                  lines.append(f"  Line {line_no}")
              if owasp_mcp:
                  lines.append(f"  OWASP MCP: {owasp_mcp}")
              lines.append(
                  f"  Details: {PIRANHA_API}/records/{ave_id}"
              )
              lines.append("")
  
      if toxic_flows:
          lines.append("TOXIC FLOWS DETECTED")
          lines.append("-" * 50)
          for flow in toxic_flows:
              title = flow.get("title", "")
              cvss = flow.get("cvss_ai", 0)
              caps = " + ".join(flow.get("capabilities", []))
              ave_ids = ", ".join(flow.get("ave_ids", []))
              lines.append(f"⛓  CRITICAL {cvss}  {title}")
              lines.append(f"  Chain: {caps}")
              lines.append(f"  AVEs: {ave_ids}")
              lines.append("")
  
      return "\n".join(lines)