Composio MCP 實戰 2026：用一把 ck_ 金鑰自動發文到 Facebook 粉專（含 6 組除錯對照表與完整可執行程式碼）

ak_  → REST /api/v3/connected_accounts      →  2 個連接
ck_  → MCP COMPOSIO_MANAGE_CONNECTIONS      →  6 個連接

① POST initialize                 → Mcp-Session-Id 在「回應標頭」裡（不是 body！）
② POST notifications/initialized
③ POST tools/list                 → 取得 meta-tool 清單
④ POST tools/call                 → 實際執行

Content-Type: application/json
Accept: application/json, text/event-stream
x-consumer-api-key: ck_xxxxxxxx
Mcp-Session-Id: <由第①步的回應標頭取得>

呼叫 COMPOSIO_MULTI_EXECUTE_TOOL
   └─ 參數裡指定 tool_slug = "FACEBOOK_CREATE_POST"
      └─ arguments 放 page_id / message / link

Step 1  確認 Facebook 連接狀態        COMPOSIO_MANAGE_CONNECTIONS (action=list)
            ↓
Step 2  列出該帳號管理的所有粉專       FACEBOOK_GET_USER_PAGES
            ↓
Step 3  確認「哪個粉專才是對的」        FACEBOOK_GET_PAGE_DETAILS（比對 website）
            ↓
Step 4  發文                          FACEBOOK_CREATE_POST
            ↓
Step 5  讀回驗證                       FACEBOOK_GET_PAGE_POSTS

call_meta("COMPOSIO_MANAGE_CONNECTIONS",
          {"toolkits": [{"name": "facebook", "action": "list"}]})

exec_tool("FACEBOOK_GET_USER_PAGES", {})

exec_tool("FACEBOOK_GET_PAGE_DETAILS", {"page_id": "<你的 page_id>"})

{
  "category": "Personal blog",
  "name": "你的粉專名稱",
  "id": "123456789012345",
  "website": "https://your-site.com/"    ← 決定性證據
}

exec_tool("FACEBOOK_CREATE_POST", {
    "page_id": "<page_id>",
    "message": "貼文文字內容",
    "link": "https://your-site.com/article/",
    "published": True,
})

{"data":{"results":[{"response":{"successful":true,
  "data":{"id":"<page_id>_<post_id>"}},"index":0}],
  "success_count":1,"error_count":0,"successful":true}}

exec_tool("FACEBOOK_GET_PAGE_POSTS", {"page_id": "<page_id>", "limit": 3})

{"successful": false, "error": "{\"status\":426,...}", "status_code": 426}

ok = ('"successful":true' in txt.replace(" ", "")
      and '"error_count":0' in txt.replace(" ", ""))

import json, urllib.request, urllib.error

CK      = "ck_xxxxxxxxxxxx"                  # consumer key（不是 ak_！）
MCP_URL = "https://connect.composio.dev/mcp"
PAGE_ID = "123456789012345"                  # 你的粉專 ID

_session = {"id": None}

def _rpc(payload):
    h = {"Content-Type": "application/json",
         "Accept": "application/json, text/event-stream",
         "x-consumer-api-key": CK}
    if _session["id"]:
        h["Mcp-Session-Id"] = _session["id"]
    req = urllib.request.Request(MCP_URL, data=json.dumps(payload).encode(),
                                 headers=h, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=180) as r:
            return r.status, dict(r.headers), r.read().decode("utf-8", "ignore")
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers), e.read().decode("utf-8", "ignore")

def _parse(body):
    """回應是 SSE 格式，必須抓 data: 行"""
    out = []
    for line in body.splitlines():
        if line.startswith("data:"):
            try: out.append(json.loads(line[5:].strip()))
            except Exception: pass
    if not out:
        try: out.append(json.loads(body))
        except Exception: pass
    return out

def connect():
    """四步握手（每次程序啟動做一次）"""
    st, hdrs, _ = _rpc({"jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {"protocolVersion": "2024-11-05", "capabilities": {},
                   "clientInfo": {"name": "my-agent", "version": "1.0"}}})
    _session["id"] = hdrs.get("Mcp-Session-Id") or hdrs.get("mcp-session-id")
    _rpc({"jsonrpc": "2.0", "method": "notifications/initialized"})

def _collect(body):
    txt = ""
    for m in _parse(body):
        if m.get("error"):
            txt += "ERROR " + json.dumps(m["error"], ensure_ascii=False)
        r = m.get("result")
        if r:
            for c in r.get("content", []):
                txt += c.get("text", "")
    return txt

def call_meta(tool_name, arguments):
    """呼叫 meta-tool（如 COMPOSIO_MANAGE_CONNECTIONS）"""
    _, _, body = _rpc({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments}})
    return _collect(body)

def exec_tool(tool_slug, arguments, rid=10):
    """執行任一 App 工具（經 MULTI_EXECUTE_TOOL 代理）"""
    _, _, body = _rpc({"jsonrpc": "2.0", "id": rid, "method": "tools/call",
        "params": {"name": "COMPOSIO_MULTI_EXECUTE_TOOL",
                   "arguments": {"tools": [{"tool_slug": tool_slug,
                                             "arguments": arguments}]}}})
    txt = _collect(body)
    ok = ('"successful":true' in txt.replace(" ", "")
          and '"error_count":0' in txt.replace(" ", ""))
    return ok, txt

# ---------------- 實際使用 ----------------
if __name__ == "__main__":
    connect()

    print("連接狀態:", call_meta("COMPOSIO_MANAGE_CONNECTIONS",
                                {"toolkits": [{"name": "facebook", "action": "list"}]})[:300])

    ok, pages = exec_tool("FACEBOOK_GET_USER_PAGES", {})
    print("粉專清單 ok =", ok)

    ok, res = exec_tool("FACEBOOK_CREATE_POST", {
        "page_id": PAGE_ID,
        "message": "貼文內容",
        "link": "https://example.com/article/",
        "published": True,
    }, rid=20)
    print("發文 ok =", ok)
    print(res[:300])

{"error":{"message":"(#200) This endpoint is deprecated since the
 required permission publish_actions is deprecated","type":"OAuthException"}}

mcp_servers:
  composio:
    url: https://connect.composio.dev/mcp
    headers:
      x-consumer-api-key: "ck_xxxxxxxx"
    timeout: 180
    connect_timeout: 60

hermes mcp add composio --url https://connect.composio.dev/mcp --auth header