test_mcp_http.py

"""MCP Streamable HTTP e2e 테스트 — 2025-03 표준 전송(단일 /mcp 엔드포인트)."""
import asyncio, os, sys, json
from datetime import timedelta
from dotenv import load_dotenv
load_dotenv()   # .env 의 MCP_TOKEN 로드
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:9000/mcp"
TO = timedelta(seconds=120)
_token = os.getenv("MCP_TOKEN", "")
HEADERS = {"Authorization": f"Bearer {_token}"} if _token else None


def _show(label, result):
    txt = result.content[0].text if result.content else "{}"
    try:
        obj = json.loads(txt)
        print(f"\n[{label}] ok={obj.get('ok')} elapsed={obj.get('elapsed_ms')}ms trace={obj.get('trace_id','')[:8]}")
        return obj
    except Exception:
        print(f"\n[{label}] {txt[:150]}")
        return {}


async def main():
    print(f"연결: {URL}  (Streamable HTTP, auth={'ON' if HEADERS else 'OFF'})")
    async with streamablehttp_client(URL, headers=HEADERS) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("=== tools/list ===")
            for t in tools.tools:
                print(f"  - {t.name}")
            r = _show("health", await session.call_tool("health", {}, read_timeout_seconds=TO))
            print(f"    status={r.get('status')} menu_count={r.get('menu_count')} engine={r.get('engine')}")
            r = _show("search_menu(SOXL 주가)", await session.call_tool(
                "search_menu", {"query": "SOXL 주가", "limit": 3, "verbose": True},
                read_timeout_seconds=TO))
            for it in r.get("results", []):
                print(f"    {it['rank']}. {it['menu_name']} | {it['menu_path']}  sim={it.get('similarity')}")
            r = _show("fetch_menu_raw()", await session.call_tool(
                "fetch_menu_raw", {}, read_timeout_seconds=TO))
            print(f"    count={r.get('count')} (첫 메뉴: {(r.get('menus') or [{}])[0].get('menu_path')})")
            print("\nMCP_HTTP_TEST_DONE")


if __name__ == "__main__":
    asyncio.run(main())