result = await session.call_tool(
                "echo",
                arguments={"name": "Hello, World!"},
                _meta={"custom_param": "custom_value"},
            )

from mcp import ClientSession, StdioServerParameters, types
class CustomClientSession(ClientSession):
    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: Any = None,
        progress_callback: Any = None,
        _meta: Dict[str, Any] | None = None,
    ) -> types.CallToolResult:
        """Send a tools/call request with optional progress callback and meta support."""

        # Properly create the Meta object if _meta is provided
        meta_obj = types.RequestParams.Meta(**_meta) if _meta else None

        # Pass meta as the 'meta' argument (not '_meta')
        params = types.CallToolRequestParams(
            name=name, arguments=arguments, _meta=meta_obj
        )

        # Build the request object
        request_obj = types.ClientRequest(
            types.CallToolRequest(
                method="tools/call",
                params=params,
            )
        )

        # Print the outgoing JSON-RPC request for debugging
        print("Outgoing JSON-RPC request:")
        print(json.dumps(request_obj.model_dump(mode="json", by_alias=True), indent=2))

        return await self.send_request(
            request_obj,
            types.CallToolResult,
            request_read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
        )
        
async def main():
    # Define server parameters
    server_params = StdioServerParameters(
        command="python",  # The command to run your server
        args=["server.py"],  # Arguments to the command
    )

    # Connect to the server
    async with stdio_client(server_params) as (read_stream, write_stream):
        # Use our custom session class
        async with CustomClientSession(read_stream, write_stream) as session:
            # Initialize the connection
            await session.initialize()

            # List available tools
            tools_result = await session.list_tools()
            print("Available tools:")
            for tool in tools_result.tools:
                print(f"  - {tool.name}: {tool.description}")

            result = await session.call_tool(
                "echo",
                arguments={"name": "Hello, World!"},
                _meta={"user_id": "1234567890"},
            )
            print(f"Stdio Echo: {result.content[0].text}")


if __name__ == "__main__":
    asyncio.run(main())

def extract_meta_from_context(ctx: Context) -> dict:
    """
    Extract meta information from the MCP context.

    Args:
        ctx: The MCP context object

    Returns:
        dict: Dictionary containing custom meta data (excluding progressToken)
    """
    if not hasattr(ctx, "request_context") or ctx.request_context is None:
        return {}

    if not hasattr(ctx.request_context, "meta") or ctx.request_context.meta is None:
        return {}

    # Convert meta object to dictionary
    meta_dict = (
        ctx.request_context.meta.model_dump()
        if hasattr(ctx.request_context.meta, "model_dump")
        else {}
    )

    return meta_dict


# Run the server
if __name__ == "__main__":
    # parser for the type of message transport channel to run the server
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--transport",
        type=str,
        default="stdio",
        choices=["stdio", "streamable-http", "websockets"],
    )
    parser.add_argument(
        "--host",
        type=str,
        default="127.0.0.1",
        help="Host to bind to (only used for SSE transport)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=8050,
        help="Port to bind to (only used for SSE transport)",
    )
    args = parser.parse_args()

    # Create an MCP server with host and port from arguments
    mcp = FastMCP(name="DemoServer", host=args.host, port=args.port)

    # Simple tool
    @mcp.tool()
    def echo(name: str, ctx: Context) -> str:
        """Echos the input back to the client

        Args:
            name: The input to echo back
        """
        # Use stderr for debugging output (won't interfere with stdio communication)
        print(f"Context: {ctx}", file=sys.stderr)
        print(f"Context type: {type(ctx)}", file=sys.stderr)
        print(f"Context content: {ctx.request_context.meta}", file=sys.stderr)

        # Use the helper function to get meta information
        meta = extract_meta_from_context(ctx)

        # Return the echo with meta information
        return f"{name}! ({meta})"

    # run the server
    if args.transport == "stdio":
        mcp.run()
    elif args.transport == "streamable-http":
        mcp.run(transport="streamable-http")
    elif args.transport == "websockets":
        raise NotImplementedError("Websockets are not supported yet")
    else:
        raise ValueError(f"Invalid transport: {args.transport}")