"""HTTP MCP 클라이언트 테스트"""
import asyncio
import httpx
from mcp import ClientSession
from mcp.client.sse import sse_client
async def test_http_mcp():
"""HTTP MCP 서버를 테스트합니다"""
server_url = "http://localhost:8000/mcp"
print(f"🔗 Connecting to MCP server: {server_url}\n")
async with sse_client(server_url) as (read, write):
async with ClientSession(read, write) as session:
# 서버 초기화
await session.initialize()
print("✅ Connected to MCP server!\n")
# 사용 가능한 Tools 목록
tools = await session.list_tools()
print(f"📋 Available Tools ({len(tools.tools)}):\n")
for i, tool in enumerate(tools.tools, 1):
print(f"{i:2}. {tool.name}")
print(f" → {tool.description}\n")
print("=" * 70 + "\n")
# Tool 테스트
print("🧪 Testing Tools:\n")
# 1. 오늘 예약 조회
print("1️⃣ get_today_reservations")
print("-" * 70)
result = await session.call_tool("get_today_reservations", {})
print(result.content[0].text)
print()
# 2. 이번 주 예약 조회
print("2️⃣ get_week_reservations")
print("-" * 70)
result = await session.call_tool("get_week_reservations", {})
print(result.content[0].text)
print()
# 3. 중복 예약 체크
print("3️⃣ check_duplicate_reservations")
print("-" * 70)
result = await session.call_tool("check_duplicate_reservations", {})
print(result.content[0].text)
print()
# 4. 매출 요약
print("4️⃣ get_revenue_summary (month)")
print("-" * 70)
result = await session.call_tool("get_revenue_summary", {"period": "month"})
print(result.content[0].text)
print()
# 5. 블랙리스트 조회
print("5️⃣ get_blacklist")
print("-" * 70)
result = await session.call_tool("get_blacklist", {})
print(result.content[0].text)
print()
print("=" * 70)
print("✅ All tests completed successfully!")
if __name__ == "__main__":
try:
asyncio.run(test_http_mcp())
except KeyboardInterrupt:
print("\n\n⚠️ Test interrupted by user")
except Exception as e:
print(f"\n\n❌ Error: {e}")
import traceback
traceback.print_exc()
