| from typing import List, Dict |
| import asyncio |
|
|
| from mcp import ClientSession, StdioServerParameters |
| from mcp.client.stdio import stdio_client |
| from mcp.client.sse import sse_client |
| from contextlib import AsyncExitStack |
|
|
| async def connect_to_stdio_server() -> None: |
| """Connect to a single MCP server.""" |
|
|
| sessions: List[ClientSession] = [] |
| exit_stack = AsyncExitStack() |
|
|
| try: |
| server_params = StdioServerParameters( |
| command="npx", |
| args=[ |
| "mcp-remote", |
| "https://jarisko-mcp-server-deployment-sample.hf.space/gradio_api/mcp/sse", |
| "--transport", |
| "sse-only" |
| ] |
| ) |
|
|
| stdio_transport = await exit_stack.enter_async_context( |
| stdio_client(server_params) |
| ) |
|
|
| read, write = stdio_transport |
| session = await exit_stack.enter_async_context( |
| ClientSession(read, write) |
| ) |
|
|
| await session.initialize() |
| sessions.append(session) |
|
|
| print("3") |
|
|
| |
| response = await session.list_tools() |
| tools = response.tools |
| print(f"{tools=}") |
|
|
| except Exception as e: |
| print(f"Failed to connect: {e}") |
|
|
|
|
| async def connect_to_sse_server(): |
|
|
| async with sse_client("https://jarisko-mcp-server-deployment-sample.hf.space/gradio_api/mcp/sse") as streams: |
| async with ClientSession(*streams) as session: |
| await session.initialize() |
| tools_result = await session.list_tools() |
| print(f"{tools_result=}") |
|
|
|
|
| asyncio.run(connect_to_sse_server()) |
