40 lines
1.7 KiB
Python
40 lines
1.7 KiB
Python
"""Minimal scripted MCP client for the Arcade eval.
|
|
|
|
Headless auth (confirmed from Arcade docs, see ../LIVE-POC.md):
|
|
Authorization: Bearer <ARCADE_API_KEY>
|
|
Arcade-User-ID: <user_id> (any stable string; an email works)
|
|
|
|
`auth_headers` is pure (no deps) so it unit-tests offline. The `mcp` SDK is imported
|
|
lazily inside the async helpers so this module loads even before deps are installed.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
|
|
def auth_headers(api_key: str, user_id: str) -> dict[str, str]:
|
|
"""Build the headless MCP auth headers for a given Arcade user_id."""
|
|
return {"Authorization": f"Bearer {api_key}", "Arcade-User-ID": user_id}
|
|
|
|
|
|
async def connect_and_list(mcp_url: str, headers: dict[str, str]) -> list[dict]:
|
|
"""Connect to an MCP gateway over streamable HTTP and return the tool list."""
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
|
|
async with streamablehttp_client(mcp_url, headers=headers) as (read, write, _):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
result = await session.list_tools()
|
|
return [t.model_dump() for t in result.tools]
|
|
|
|
|
|
async def call_tool(mcp_url: str, headers: dict[str, str], name: str, args: dict) -> dict:
|
|
"""Invoke a single tool through an MCP gateway and return the result payload."""
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
|
|
async with streamablehttp_client(mcp_url, headers=headers) as (read, write, _):
|
|
async with ClientSession(read, write) as session:
|
|
await session.initialize()
|
|
result = await session.call_tool(name, args)
|
|
return result.model_dump()
|