apex-mcp-wrapper

name	apex-mcp-wrapper
description	Use when building a standalone FastMCP server that wraps a REST API as MCP tools — either adding/editing a tool in APEX's own `mcp/server.py`, or scaffolding a NEW wrapper for an external vendor API (CyberArk, ServiceNow, etc.). Triggers on "add MCP tool", "wrap endpoint as MCP", "sync mcp/server.py", "MCP wrapper for {vendor}", "expose REST API as MCP".

APEX MCP Wrapper (writing & extending)

Overview

A wrapper is a FastMCP server that exposes a REST API as MCP tools — one async @mcp.tool() function per endpoint. An MCP client calls a tool, the wrapper calls REST, returns the result as a string. Two situations:

• A. APEX's own wrapper (mcp/server.py) — mirrors the APEX REST API.
• B. A new external API (mcp-<vendor>/server.py) — wraps a third-party API for the agent runtime. Same FastMCP shape, but you write the auth + base-URL + HTTP plumbing yourself, and it usually differs from APEX's static-key model.

This is NOT the backend integration layer — to make a wrapper callable by the agent, see apex-mcp-integration.

Quick reference

Thing	Value
Library	from mcp.server.fastmcp import FastMCP (mcp>=1.26.0)
HTTP client	httpx (async, 30s timeout)
Transport	streamable-http if APEX's agent consumes it (mcp.run(transport="streamable-http") + /health) — APEX connects over HTTP, the stdio+command path is legacy (L199). stdio ONLY for a server an external client spawns itself (e.g. mcp/server.py in Claude Desktop).
Tool unit	one @mcp.tool() async function, returns str (wrap in _fmt())
All config	env vars, read once at module top
Helpers	_request() (auth + timeout + 204 + errors), _fmt() — never re-roll per tool

A. Add ONE tool to APEX's mcp/server.py

HARD RULE (CLAUDE.md): every new backend endpoint MUST get a matching tool here, same PR.

@mcp.tool()
async def get_bot(bot_id: str) -> str:
    """Fetch a single bot by UUID. Returns the bot record as JSON."""
    return _fmt(await _request("GET", f"/api/bots/{bot_id}"))

APEX auth lives in _headers() (picks APEX_API_KEY vs APEX_JWT from APEX_AUTH_MODE). Reuse the existing helpers; don't open your own httpx client.

B. Scaffold a NEW external-API wrapper

Create mcp-<vendor>/ with server.py + requirements.txt + README.md. Drive everything off env vars (base URL, auth, TLS) — never hardcode a host or credential.

Pick the auth shape — this is where external APIs diverge from APEX:

Auth shape	Examples	Pattern
Static token, Bearer	APEX, GitHub PAT	header Authorization: Bearer {token}
Static token, raw (no Bearer)	CyberArk session token, some appliances	header Authorization: {token}
Session-based (login → token → reuse)	CyberArk Logon, many on-prem appliances	logon once, cache token, re-logon + retry on 401
API key in custom header	Figma X-Figma-Token	service-specific header name

Session-based auth + query params + key aliasing — the full shape (from the CyberArk wrapper):

BASE_URL = (os.environ.get("CYBERARK_BASE_URL") or "").rstrip("/")
VERIFY_SSL = (os.environ.get("CYBERARK_VERIFY_SSL") or "true").lower() == "true"  # self-signed → false
_token: str | None = None
_token_lock = asyncio.Lock()                      # stop concurrent double-logon

async def _request(method, path, *, body=None, params=None):
    global _token
    clean = {k: v for k, v in (params or {}).items() if v is not None}   # drop unset query params
    async with httpx.AsyncClient(verify=VERIFY_SSL, timeout=30.0) as c:
        async def _send(tok):
            return await c.request(method, f"{BASE_URL}{path}", json=body, params=clean or None,
                                   headers={"Authorization": tok})        # raw token, NOT Bearer
        r = await _send(await _ensure_token())
        if r.status_code == 401:                  # token expired → re-logon once, retry
            async with _token_lock: _token = None
            r = await _send(await _ensure_token())
    if r.status_code == 204: return {"status": "ok"}
    if not r.is_success: return {"error": r.status_code, "detail": r.text}
    try: return r.json()
    except Exception: return r.text

@mcp.tool()
async def list_accounts(search: str | None = None, safe_filter: str | None = None,
                        limit: int | None = None) -> str:
    """List accounts. safe_filter e.g. "safeName eq ProdSafe"."""
    return _fmt(await _request("GET", "/API/Accounts",
                               params={"search": search, "filter": safe_filter, "limit": limit}))

@mcp.tool()
async def add_account(platform_id: str, safe_name: str, user_name: str | None = None) -> str:
    """Onboard an account. platform_id is case-sensitive."""
    body = {"platformId": platform_id, "safeName": safe_name}   # snake_case arg → exact API key
    if user_name is not None: body["userName"] = user_name
    return _fmt(await _request("POST", "/API/Accounts", body=body))

Rules that hold for BOTH A and B:

• Always return str via _fmt(...) — never a raw dict/list.
• Auth/timeout/error handling live in _request() — one helper, every tool calls it.
• Optional params default to None, added to body/params only when set.
• Docstring = tool description the LLM caller reads — say what it does + what it returns.
• Key aliasing is fine when the API forces it. APEX has clean keys (1:1, no aliasing). External APIs often use PascalCase/inconsistent keys — keep tool params snake_case and map to the exact API key inside the function. Don't expose ugly keys to the caller.
• PATCH/PUT bodies can be a list (JSON Patch [{op,path,value}]) or a replacement object — _request passes body straight to json=, so either works.

Common mistakes

Mistake	Fix
Returning a dict	Wrap in _fmt() — clients expect a string
New httpx.AsyncClient() inside a tool	Use _request(); it centralizes auth + timeout + 204 + errors
Sending Bearer {token} when the API wants the raw token	Check the vendor docs — CyberArk uses the bare token
Treating a session API like a static key	Cache the token, re-logon + retry on 401, lock against double-logon
Hardcoding the host or a credential	All config via env, read once at module top
GET filters jammed into the path	Pass them as params= and drop Nones
verify=True against a self-signed appliance	Expose *_VERIFY_SSL env (default true, operator sets false)
APEX endpoint shipped without a wrapper tool	HARD RULE violation — add it in the same PR

Verify after building

python3 -c "import asyncio, server; \
  t=asyncio.run(server.mcp.list_tools()); \
  print(len(t),'tools'); assert all(x.description for x in t)"

Confirms imports, decorator registration, unique names, and docstrings. Then smoke-call one tool from an MCP client (needs a reachable host for an external wrapper).