Client Transports

from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

http_url = "http://localhost:8000/mcp"

# Option 1: Inferred transport (default for HTTP URLs)
client_inferred = Client(http_url)

# Option 2: Explicit transport (e.g., to add custom headers)
headers = {"Authorization": "Bearer mytoken"}
transport_explicit = StreamableHttpTransport(url=http_url, headers=headers)
client_explicit = Client(transport_explicit)

async def use_streamable_http_client(client):
    async with client:
        tools = await client.list_tools()
        print(f"Connected via Streamable HTTP, found tools: {tools}")

# asyncio.run(use_streamable_http_client(client_inferred))
# asyncio.run(use_streamable_http_client(client_explicit))

from fastmcp import Client
from fastmcp.client.transports import SSETransport

sse_url = "http://localhost:8000/sse"

# Since v2.3.0, HTTP URLs default to StreamableHttpTransport,
# so you must explicitly use SSETransport for SSE connections
transport_explicit = SSETransport(url=sse_url)
client_explicit = Client(transport_explicit)

async def use_sse_client(client):
    async with client:
        tools = await client.list_tools()
        print(f"Connected via SSE, found tools: {tools}")

# asyncio.run(use_sse_client(client_explicit))

from fastmcp import Client
from fastmcp.client.transports import PythonStdioTransport

server_script = "my_mcp_server.py" # Assumes this file exists and runs mcp.run()

# Option 1: Inferred transport
client_inferred = Client(server_script)

# Option 2: Explicit transport (e.g., to use a specific python executable or add args)
transport_explicit = PythonStdioTransport(
    script_path=server_script,
    python_cmd="/usr/bin/python3.11", # Specify python version
    # args=["--some-server-arg"], # Pass args to the script
    # env={"MY_VAR": "value"},   # Set environment variables
    # cwd="/path/to/run/in"       # Set working directory
)
client_explicit = Client(transport_explicit)

async def use_stdio_client(client):
    async with client:
        tools = await client.list_tools()
        print(f"Connected via Python Stdio, found tools: {tools}")

# asyncio.run(use_stdio_client(client_inferred))
# asyncio.run(use_stdio_client(client_explicit))

from fastmcp import Client
from fastmcp.client.transports import NodeStdioTransport

node_server_script = "my_mcp_server.js" # Assumes this JS file starts an MCP server on stdio

# Option 1: Inferred transport
client_inferred = Client(node_server_script)

# Option 2: Explicit transport
transport_explicit = NodeStdioTransport(
    script_path=node_server_script,
    node_cmd="node" # Or specify path to Node executable
)
client_explicit = Client(transport_explicit)

# Usage is the same as other clients
# async with client_explicit:
#    tools = await client_explicit.list_tools()

from fastmcp.client.transports import UvxStdioTransport

# Example: Run a hypothetical 'cloud-analyzer-mcp' tool via uvx
# Assume this tool, when run, starts an MCP server on stdio
transport = UvxStdioTransport(
    tool_name="cloud-analyzer-mcp",
    # from_package="cloud-analyzer-cli", # Optionally specify package if tool name differs
    # with_packages=["boto3", "requests"], # Add dependencies if needed
    # tool_args=["--config", "prod.yaml"] # Pass args to the tool itself
)
client = Client(transport)

# async with client:
#     analysis = await client.call_tool("analyze_bucket", {"name": "my-data"})

from fastmcp.client.transports import NpxStdioTransport

# Example: Run a hypothetical 'npm-mcp-server-package' via npx
transport = NpxStdioTransport(
    package="npm-mcp-server-package",
    # args=["--port", "stdio"] # Args passed to the package script
)
client = Client(transport)

# async with client:
#     response = await client.call_tool("get_npm_data", {})

from fastmcp import FastMCP, Client
from fastmcp.client.transports import FastMCPTransport

# 1. Create your FastMCP server instance
server = FastMCP(name="InMemoryServer")
@server.tool()
def ping(): return "pong"

# 2. Create a client pointing directly to the server instance
# Option A: Inferred
client_inferred = Client(server)

# Option B: Explicit
transport_explicit = FastMCPTransport(mcp=server)
client_explicit = Client(transport_explicit)

# 3. Use the client (no subprocess or network involved)
async def test_in_memory():
    async with client_inferred: # Or client_explicit
        result = await client_inferred.call_tool("ping")
        print(f"In-memory call result: {result[0].text}") # Output: pong

# asyncio.run(test_in_memory())