Spaces:
Paused
Paused
File size: 1,923 Bytes
a5784e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | import asyncio
import json
import os
from typing import Any, Dict
import httpx
def _normalize_endpoint(ep: str) -> str:
if not ep:
raise RuntimeError("MCP HTTP endpoint not provided")
return ep.rstrip("/")
async def execute_mcp_tool(name: str, params: Dict[str, Any]) -> str:
"""
Minimal MCP-over-HTTP adapter:
- POST {MCP_HTTP_ENDPOINT}/tools/execute with {name, arguments}
- Returns JSON string.
Compatible with servers exposing MCP-like HTTP interface.
"""
ep = os.environ.get("MCP_HTTP_ENDPOINT")
if not ep:
raise RuntimeError("MCP_HTTP_ENDPOINT not configured")
url = f"{_normalize_endpoint(ep)}/tools/execute"
payload = {"name": name, "arguments": params}
headers = {"Content-Type": "application/json"}
timeout = float(os.environ.get("MCP_HTTP_TIMEOUT", "15"))
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(url, json=payload, headers=headers)
resp.raise_for_status()
try:
data = resp.json()
except asyncio.CancelledError:
raise
except Exception:
data = {"raw": resp.text}
return json.dumps(data, ensure_ascii=False)
async def execute_mcp_tool_with_endpoint(
endpoint: str, name: str, params: Dict[str, Any]
) -> str:
url = f"{_normalize_endpoint(endpoint)}/tools/execute"
payload = {"name": name, "arguments": params}
headers = {"Content-Type": "application/json"}
timeout = float(os.environ.get("MCP_HTTP_TIMEOUT", "15"))
async with httpx.AsyncClient(timeout=timeout) as client:
resp = await client.post(url, json=payload, headers=headers)
resp.raise_for_status()
try:
data = resp.json()
except asyncio.CancelledError:
raise
except Exception:
data = {"raw": resp.text}
return json.dumps(data, ensure_ascii=False)
|