Ruby MCP Client

# Set filesystem scope boundaries client.roots = [ { uri: 'file:///home/user/project', name: 'Project' }, { uri: 'file:///var/log', name: 'Logs' } ] # Access current roots client.roots

Sampling (Server-requested LLM completions)

# Configure handler when creating client client = MCPClient.connect('http://server/mcp', sampling_handler: ->(messages, model_prefs, system_prompt, max_tokens) { # Process server's LLM request { 'model' => 'gpt-4', 'stopReason' => 'endTurn', 'role' => 'assistant', 'content' => { 'type' => 'text', 'text' => 'Response here' } } } )

Sampling tool calling (SEP-1577) is opt-in: passsampling_supports_tools: trueto declare thesampling.toolscapability. The handler then receives the full request params (includingtools/toolChoice) as an optional fifth argument; without the opt-in, tool-enabled sampling requests are rejected with-32602as the spec requires:

client = MCPClient::Client.new( mcp_server_configs: [...], sampling_supports_tools: true, sampling_handler: ->(messages, prefs, system_prompt, max_tokens, params = nil) { tools = params && params['tools'] # ToolUseContent may be returned in content # ... } )

Attach a per-call progress callback — the client generates a uniqueprogressToken, places it in the request_meta, and routes matchingnotifications/progressto your block while the request is active (stale tokens after completion are dropped):

client.call_tool('long_running', args, progress: ->(progress, total, message) { puts "#{message}: #{progress}/#{total}" })

A request-level_meta(e.g. a hand-pickedprogressToken) can also be passed inside the arguments under the'_meta'key on every transport — it is hoisted to the JSON-RPC params level on the wire, never sent as a tool argument.

Timeouts are configurable per request in addition to the per-serverread_timeout. A timed-out request raisesMCPClient::Errors::RequestTimeoutError(aTransportErrorsubclass), isneversilently re-sent by the retry layer, and a best-effortnotifications/cancelledis sent for the abandoned request (never forinitialize, and task-augmented calls usetasks/cancelinstead):

client.send_rpc('tools/call', params: { name: 'slow', arguments: {} }, timeout: 300) server.rpc_request('tools/list', {}, timeout: 5)

Hosts can present their ownImplementationinfo (sent asclientInfoduring initialize;nameandversionrequired —title,description,websiteUrl,iconsoptional), and read the server'sinstructionshint after connecting:

client = MCPClient::Client.new( mcp_server_configs: [...], client_info: { 'name' => 'my-ide', 'version' => '2.0.0', 'description' => 'An MCP-powered IDE' } ) client.servers.first.connect puts client.servers.first.instructions # e.g. "Use the search tool before answering."

result = client.complete( ref: { type: 'ref/prompt', name: 'greeting' }, argument: { name: 'name', value: 'A' } ) # => { 'values' => ['Alice', 'Alex'], 'total' => 100, 'hasMore' => true }

# Set log level client.log_level = 'debug' # debug/info/notice/warning/error/critical # Handle log notifications client.on_notification do |server, method, params| if method == 'notifications/message' puts "[#{params['level']}] #{params['logger']}: #{params['data']}" end end

Tasks (Long-running, task-augmented tools)

A task-capable server (one advertisingtasks.requests.tools.call) can run a tool whoseexecution.taskSupportisoptionalorrequiredas a background task: the call returns immediately with a task handle, and the result is fetched later. Try it locally:python3 examples/echo_server_streamable.py &then./examples/tasks_example.rbruns the full lifecycle against a task-capable demo server.

tool = client.find_tool('long_job') tool.supports_task? # execution.taskSupport is optional/required? # Create the task (returns immediately); ttl is the requested lifetime in ms task = client.call_tool_as_task('long_job', { input: 'data' }, ttl: 60_000) # Poll until the task reaches a terminal (or input-required) status, # honoring the server's suggested poll interval until task.terminal? || task.input_required? sleep((task.poll_interval || 1000) / 1000.0) task = client.get_task(task) # tasks/get, routed to the task's own server end # Retrieve the underlying result (e.g. a CallToolResult) via tasks/result result = client.get_task_result(task) # List and cancel tasks page = client.list_tasks # { tasks: [...], next_cursor: ... } client.cancel_task(task) # tasks/cancel

Task IDs are only unique within the server that issued them, so pass theTaskreturned bycall_tool_as_task— it carries its own server. A bare task ID also works when the client has a single server; with several servers configured it raisesArgumentErrorrather than guessing, so name the server explicitly:

client.get_task('task-123', server: 'my-server') # React to server-pushed status updates client.on_notification do |server, method, params| puts "Task #{params['taskId']} -> #{params['status']}" if method == 'notifications/tasks/status' end

Elicitation (Server-initiated user interactions)

client = MCPClient::Client.new( mcp_server_configs: [MCPClient.stdio_config(command: 'python server.py')], elicitation_handler: ->(message, schema) { puts "Server asks: #{message}" # Return: { 'action' => 'accept', 'content' => { 'field' => 'value' } } # Or: { 'action' => 'decline' } or { 'action' => 'cancel' } } )

For more control, usecreate_clientwith explicit configs:

client = MCPClient.create_client( mcp_server_configs: [ MCPClient.stdio_config(command: 'npx server', name: 'local'), MCPClient.sse_config( base_url: 'https://api.example.com/sse', headers: { 'Authorization' => 'Bearer TOKEN' }, read_timeout: 30, ping: 10, retries: 3 ), MCPClient.http_config( base_url: 'https://api.example.com', endpoint: '/rpc', headers: { 'Authorization' => 'Bearer TOKEN' } ), MCPClient.streamable_http_config( base_url: 'https://api.example.com/mcp', read_timeout: 60, retries: 3 ) ], logger: Logger.new($stdout) ) # Or load from JSON file client = MCPClient.create_client(server_definition_file: 'servers.json')

Theretries:option controls automatic retry with exponential backoff. Only failures where the request most likely didnotcomplete at the server are retried: transport/network errors and HTTP5xxresponses. Application-level failures — a JSON-RPC error response or an HTTP4xx— areneverretried, because the server already processed or rejected the request. Retryable server failures raiseMCPClient::Errors::TransientServerError, a subclass ofMCPClient::Errors::ServerError, so existingrescue ServerErrorhandlers are unaffected.

begin client.call_tool('send_invoice', { customer: 'acme' }) rescue MCPClient::Errors::TransportError => e # The server may or may not have sent the invoice. Check before retrying. end

The same reasoning excludesRequestTimeoutErrorandResponseTooLargeErrorfrom retries, and applies to session recovery: if atools/callcomes back with an expired-session 404, the client starts a fresh session but doesnotre-send the call — it raises so you can decide. Idempotent requests are re-sent against the new session as before.

A gzip-encoded response is decompressed incrementally and abandoned once it expands pastmax_decompressed_body_bytes(default64 MiB), so a small highly-compressed body cannot exhaust memory. Exceeding it raisesMCPClient::Errors::ResponseTooLargeError.

Raise the limit if you legitimately exchange very large payloads — base64 resource blobs or audio — so that whether a response is accepted does not depend on the server's choice to compress it:

MCPClient.http_config(base_url: 'https://internal.company.com') do |faraday| faraday.ssl.cert_store = custom_cert_store faraday.ssl.verify = true end

{ "mcpServers": { "filesystem": { "type": "stdio", "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home"] }, "api": { "type": "streamable_http", "url": "https://api.example.com/mcp", "headers": { "Authorization": "Bearer TOKEN" } } } }

require 'mcp_client' require 'openai' mcp = MCPClient.connect('npx -y @modelcontextprotocol/server-filesystem .') tools = mcp.to_openai_tools client = OpenAI::Client.new(api_key: ENV['OPENAI_API_KEY']) response = client.chat.completions.create( model: 'gpt-4', messages: [{ role: 'user', content: 'List files' }], tools: tools )

require 'mcp_client' require 'anthropic' mcp = MCPClient.connect('npx -y @modelcontextprotocol/server-filesystem .') tools = mcp.to_anthropic_tools client = Anthropic::Client.new(access_token: ENV['ANTHROPIC_API_KEY']) # Use tools with Claude API

require 'mcp_client' require 'ruby_llm' RubyLLM.configure { |c| c.openai_api_key = ENV['OPENAI_API_KEY'] } mcp = MCPClient.connect('http://localhost:8931/mcp') # Playwright MCP # Wrap each MCP tool as a RubyLLM tool tools = mcp.list_tools.map do |t| tool_name = t.name Class.new(RubyLLM::Tool) do description t.description params t.schema define_method(:name) { tool_name } define_method(:execute) { |args| mcp.call_tool(tool_name, args) } end.new end chat = RubyLLM.chat(model: 'gpt-4o-mini') tools.each { |tool| chat.with_tool(tool) } response = chat.ask('Navigate to google.com and tell me the page title')

Runbundle installfirst. The script preflight-checks the following and prints a warning (it doesnotabort) for anything missing; affected examples are then skipped or fail:

examples/run_all_examples.sh # run everything runnable on this machine RUN_AI=0 examples/run_all_examples.sh # skip the paid-LLM examples RUN_NPX=0 examples/run_all_examples.sh # skip the npx-based example (json_input) LOG_DIR=/path examples/run_all_examples.sh # write logs to a chosen dir PYTHON=python3.12 TIMEOUT=180 examples/run_all_examples.sh # override interpreter and per-example timeout

Real secrets live inexamples/secrets.env, which isgitignoredand sourced automatically (everyKEY=valueline is exported) when present. Copy the tracked template to get started:

cp examples/secrets.env.example examples/secrets.env # then set ZAPIER_MCP_TOKEN=... to enable the Zapier streamable-HTTP example

require 'mcp_client' require 'mcp_client/auth/browser_oauth' oauth = MCPClient::Auth::OAuthProvider.new( server_url: 'https://api.example.com/mcp', redirect_uri: 'http://localhost:8080/callback', scope: 'mcp:read mcp:write' ) browser_oauth = MCPClient::Auth::BrowserOAuth.new(oauth) token = browser_oauth.authenticate # Opens browser, handles callback client = MCPClient::Client.new( mcp_server_configs: [{ type: 'streamable_http', base_url: 'https://api.example.com/mcp', oauth_provider: oauth }] )

- Client ID Metadata Documents (SEP-991)— passclient_id_metadata_url: 'https://myapp.example/oauth-client.json'(an HTTPS URL with a path, which doubles as theclient_id); when the authorization server advertisesclient_id_metadata_document_supported, dynamic client registration is skipped entirely.
- Scope challenges (SEP-835)— an HTTP 403insufficient_scopechallenge raisesMCPClient::Errors::InsufficientScopeError(aConnectionErrorsubclass) exposing#scopeand#error_description; the challenged scopes are treated as authoritative for the next authorization flow.
- PKCE— authorization refuses to proceed when the authorization server does not advertisecode_challenge_methods_supportedincludingS256.

client.on_notification do |server, method, params| case method when 'notifications/tools/list_changed' client.clear_cache # Auto-handled when 'notifications/message' puts "Log: #{params['data']}" when 'notifications/roots/list_changed' puts "Roots changed" end end

- Session capture: ExtractsMcp-Session-Idfrom initialize response
- Session persistence: Includes session header in subsequent requests
- Session termination: Sends DELETE request during cleanup
- Resumability(Streamable HTTP, SEP-1699): tracks SSE event IDs and, when a response stream is interrupted, resumes via GET withLast-Event-IDso the server can replay missed messages — honoring the server'sretry:directive

# Start server python examples/echo_server_streamable.py

# Connect and use client = MCPClient.connect('http://localhost:8931/mcp') tools = client.list_tools result = client.call_tool('echo', { message: 'Hello!' })