31 lines
870 B
Python
31 lines
870 B
Python
"""Manual integration test for the streamable HTTP MCP client."""
|
|
|
|
import argparse
|
|
import asyncio
|
|
|
|
from mcp.client.session import ClientSession
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
|
|
|
|
async def main(url: str) -> None:
|
|
"""Connect to a streamable HTTP MCP server and initialize a session."""
|
|
async with (
|
|
streamable_http_client(url) as streams,
|
|
ClientSession(streams[0], streams[1]) as session,
|
|
):
|
|
await session.initialize()
|
|
print("Client initialized!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(
|
|
description="Test client for the MCP streamable HTTP transport"
|
|
)
|
|
parser.add_argument(
|
|
"--url",
|
|
default="http://localhost:8000/mcp",
|
|
help="MCP streamable HTTP endpoint URL",
|
|
)
|
|
args = parser.parse_args()
|
|
asyncio.run(main(args.url))
|