38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
import asyncio
|
|
import os
|
|
|
|
|
|
def main():
|
|
KEY = open("/home/ubuntu/quantux-mcp/.env").read().split("MCP_API_KEY=")[1].splitlines()[0].strip()
|
|
URL = os.environ.get("MCP_URL", "http://127.0.0.1:8091/mcp")
|
|
|
|
import httpx2
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
|
|
async def run():
|
|
http_client = httpx2.AsyncClient(
|
|
headers={"Authorization": f"Bearer {KEY}"}
|
|
)
|
|
async with streamable_http_client(URL, http_client=http_client) as (read, write):
|
|
async with ClientSession(read, write) as session:
|
|
init = await session.initialize()
|
|
print("server:", init.server_info.name, init.server_info.version)
|
|
print("protocol:", init.protocol_version)
|
|
|
|
tools = await session.list_tools()
|
|
print("tools count:", len(tools.tools))
|
|
print("tool names:", [t.name for t in tools.tools])
|
|
|
|
r = await session.call_tool("quantux_health", {})
|
|
print("health ->", r.content[0].text)
|
|
|
|
r = await session.call_tool("quantux_list_apps", {})
|
|
print("apps ->", r.content[0].text)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|