test

app.py

@@ -1,83 +1,29 @@
 """
-cd to the `examples/snippets/clients` directory and run:
-    uv run client
+Run from the repository root:
+    uv run examples/snippets/clients/streamable_basic.py
 """
 
 import asyncio
-import os
 
-from pydantic import AnyUrl
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
 
-from mcp import ClientSession, StdioServerParameters, types
-from mcp.client.stdio import stdio_client
-from mcp.shared.context import RequestContext
 
-# Create server parameters for stdio connection
-server_params = StdioServerParameters(
-    command="uv",  # Using uv to run the server
-    args=["run", "server", "fastmcp_quickstart", "stdio"],  # We're already in snippets dir
-    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
-)
-
-
-# Optional: create a sampling callback
-async def handle_sampling_message(
-    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
-) -> types.CreateMessageResult:
-    print(f"Sampling request: {params.messages}")
-    return types.CreateMessageResult(
-        role="assistant",
-        content=types.TextContent(
-            type="text",
-            text="Hello, world! from model",
-        ),
-        model="gpt-3.5-turbo",
-        stopReason="endTurn",
-    )
-
-
-async def run():
-    async with stdio_client(server_params) as (read, write):
-        async with ClientSession(read, write, sampling_callback=handle_sampling_message) as session:
+async def main():
+    # Connect to a streamable HTTP server
+    async with streamablehttp_client("http://localhost:8000/mcp") as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        # Create a session using the client streams
+        async with ClientSession(read_stream, write_stream) as session:
             # Initialize the connection
             await session.initialize()
-
-            # List available prompts
-            prompts = await session.list_prompts()
-            print(f"Available prompts: {[p.name for p in prompts.prompts]}")
-
-            # Get a prompt (greet_user prompt from fastmcp_quickstart)
-            if prompts.prompts:
-                prompt = await session.get_prompt("greet_user", arguments={"name": "Alice", "style": "friendly"})
-                print(f"Prompt result: {prompt.messages[0].content}")
-
-            # List available resources
-            resources = await session.list_resources()
-            print(f"Available resources: {[r.uri for r in resources.resources]}")
-
             # List available tools
             tools = await session.list_tools()
-            print(f"Available tools: {[t.name for t in tools.tools]}")
-
-            # Read a resource (greeting resource from fastmcp_quickstart)
-            resource_content = await session.read_resource(AnyUrl("greeting://World"))
-            content_block = resource_content.contents[0]
-            if isinstance(content_block, types.TextContent):
-                print(f"Resource content: {content_block.text}")
-
-            # Call a tool (add tool from fastmcp_quickstart)
-            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
-            result_unstructured = result.content[0]
-            if isinstance(result_unstructured, types.TextContent):
-                print(f"Tool result: {result_unstructured.text}")
-            result_structured = result.structuredContent
-            print(f"Structured tool result: {result_structured}")
-
-
-def main():
-    """Entry point for the client script."""
-    asyncio.run(run())
+            print(f"Available tools: {[tool.name for tool in tools.tools]}")
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())