diff --git a/netx_api/mcp_remote/asgi_client.py b/netx_api/mcp_remote/asgi_client.py index 56eaa95..ec1ff48 100644 --- a/netx_api/mcp_remote/asgi_client.py +++ b/netx_api/mcp_remote/asgi_client.py @@ -2,6 +2,8 @@ from __future__ import annotations +import asyncio +import contextvars import json import os import re @@ -141,21 +143,43 @@ def _default_timeout() -> float: return 120.0 -def _client(timeout: float) -> httpx.Client: - app = get_app() +_MAIN_LOOP: asyncio.AbstractEventLoop | None = None + + +def bind_main_loop(loop: asyncio.AbstractEventLoop) -> None: + """Remember the API loop; tool handlers run in worker threads and submit ASGI calls to it.""" + global _MAIN_LOOP + _MAIN_LOOP = loop + + +def _request(app: Any, method: str, path: str, *, timeout: float, **kwargs: Any) -> httpx.Response: if app is None: base = ( os.getenv("NETX_API_URL") or os.getenv("OCLAW_NETX_BASE_URL") or "http://127.0.0.1:8890" ).rstrip("/") - return httpx.Client(base_url=base, timeout=timeout, trust_env=False) - return httpx.Client( - transport=httpx.ASGITransport(app=app), - base_url="http://mcp.internal", - timeout=timeout, - trust_env=False, - ) + with httpx.Client(base_url=base, timeout=timeout, trust_env=False) as client: + return client.request(method, path, **kwargs) + + # httpx.ASGITransport is async-only: run the in-process call as a coroutine. + async def _go() -> httpx.Response: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://mcp.internal", + timeout=timeout, + trust_env=False, + ) as client: + return await client.request(method, path, **kwargs) + + loop = _MAIN_LOOP + if loop is not None and loop.is_running(): + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run_coroutine_threadsafe(_go(), loop).result(timeout + 5) + raise RuntimeError("http_json must not be called on the event loop thread") + return asyncio.run(_go()) def http_json( @@ -183,29 +207,30 @@ def http_json( if tok: headers["authorization"] = f"Bearer {tok}" try: - with _client(to) as client: - resp = client.request( - method, - path, - params=merged or None, - json=body, - headers=headers, - ) - text = resp.text - if not resp.is_success: - return { - "ok": False, - "error": f"netx_http_{resp.status_code}", - "detail": text[:800], - "hint": "检查 API 地址、令牌 scope 和资源 ID。", - } - data = resp.json() if text else {} - if isinstance(data, dict): - data = localize_payload(data) - data = redact(data, tokens=(tok,)) - else: - data = {"raw": data} - return {"ok": True, "data": data} + resp = _request( + get_app(), + method, + path, + timeout=to, + params=merged or None, + json=body, + headers=headers, + ) + text = resp.text + if not resp.is_success: + return { + "ok": False, + "error": f"netx_http_{resp.status_code}", + "detail": text[:800], + "hint": "检查 API 地址、令牌 scope 和资源 ID。", + } + data = resp.json() if text else {} + if isinstance(data, dict): + data = localize_payload(data) + data = redact(data, tokens=(tok,)) + else: + data = {"raw": data} + return {"ok": True, "data": data} except Exception as exc: return { "ok": False, @@ -257,7 +282,8 @@ def http_json_many( return idx, out with ThreadPoolExecutor(max_workers=workers) as pool: - futs = [pool.submit(_one, i, r) for i, r in enumerate(requests)] + # Copy the caller's context so each request keeps the caller's token (RBAC). + futs = [pool.submit(contextvars.copy_context().run, _one, i, r) for i, r in enumerate(requests)] for fut in as_completed(futs): idx, out = fut.result() results[idx] = out diff --git a/netx_api/mcp_remote/server.py b/netx_api/mcp_remote/server.py index 719701b..28f4309 100644 --- a/netx_api/mcp_remote/server.py +++ b/netx_api/mcp_remote/server.py @@ -77,8 +77,13 @@ async def _on_call_tool( try: # Handlers are sync and may nest ASGI calls; run off the event loop thread # so BaseHTTPMiddleware on the outer MCP request cannot deadlock. + import asyncio + import anyio + from .asgi_client import bind_main_loop + + bind_main_loop(asyncio.get_running_loop()) result = await anyio.to_thread.run_sync(call_tool, name, dict(args)) except Exception as exc: result = { diff --git a/tests/mcp_remote/test_mcp_tcp_framing.py b/tests/mcp_remote/test_mcp_tcp_framing.py index 68efa1e..60dab21 100644 --- a/tests/mcp_remote/test_mcp_tcp_framing.py +++ b/tests/mcp_remote/test_mcp_tcp_framing.py @@ -187,3 +187,28 @@ def test_mcp_tcp_httpx_initialize(live_mcp_port: int): assert r.status_code == 200, (r.status_code, r.text[:300]) data = r.json() assert data["result"]["serverInfo"]["name"] == "netx" + + +def test_mcp_tcp_tool_call_reaches_api(live_mcp_port: int): + """tools/call must run the in-process API request (not just list tools).""" + body = { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": {"name": "getTopologyTree", "arguments": {}}, + } + r = httpx.post( + f"http://127.0.0.1:{live_mcp_port}/mcp", + json=body, + headers={ + "Accept": "application/json, text/event-stream", + "MCP-Protocol-Version": "2024-11-05", + }, + timeout=60.0, + trust_env=False, + ) + assert r.status_code == 200, (r.status_code, r.text[:300]) + result = r.json()["result"] + text = result["content"][0]["text"] + assert "netx_request_failed" not in text, text[:300] + assert not result.get("isError"), text[:300]