read_resource
Retrieve content and metadata from a specific resource by providing its URI. This tool reads resources from connected MCP servers for testing and validation purposes.
Instructions
Read a specific resource from the connected MCP server.
Reads a resource by URI and returns its content along with metadata.
Returns: Dictionary with resource content including: - success: True if resource was read successfully - resource: Object with uri, mimeType, and content - metadata: Content size and request timing
Raises: Returns error dict for various failure scenarios: - not_connected: No active connection - resource_not_found: Resource doesn't exist on server - execution_error: Resource read failed
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| uri | Yes | URI of the resource to read (e.g., 'config://settings') |
Implementation Reference
- The core handler implementation for the 'read_resource' MCP tool. Decorated with @mcp.tool for automatic registration. Handles connection verification, resource reading via FastMCP client, content extraction (text or base64-encoded blob), comprehensive error handling (not connected, not found, etc.), progress updates via ctx, logging, statistics, and returns structured JSON-like dict with success, resource data, and metadata.
@mcp.tool async def read_resource( uri: Annotated[str, "URI of the resource to read (e.g., 'config://settings')"], ctx: Context ) -> dict[str, Any]: """Read a specific resource from the connected MCP server. Reads a resource by URI and returns its content along with metadata. Returns: Dictionary with resource content including: - success: True if resource was read successfully - resource: Object with uri, mimeType, and content - metadata: Content size and request timing Raises: Returns error dict for various failure scenarios: - not_connected: No active connection - resource_not_found: Resource doesn't exist on server - execution_error: Resource read failed """ start_time = time.perf_counter() try: # Verify connection exists client, state = ConnectionManager.require_connection() # User-facing progress update await ctx.info(f"Reading resource '{uri}' from server") # Detailed technical log logger.info( f"Reading resource '{uri}' from server", extra={"resource_uri": uri}, ) # Read the resource # Note: FastMCP Client's read_resource() returns a list directly, not a ReadResourceResult resource_start = time.perf_counter() contents_list = await client.read_resource(uri) resource_elapsed_ms = (time.perf_counter() - resource_start) * 1000 # Increment statistics ConnectionManager.increment_stat("resources_accessed") total_elapsed_ms = (time.perf_counter() - start_time) * 1000 # Extract resource content # FastMCP returns list[TextResourceContents | BlobResourceContents] directly content = None mime_type = None content_size = 0 if isinstance(contents_list, list) and len(contents_list) > 0: content_item = contents_list[0] # Check for different content types if hasattr(content_item, "text"): content = content_item.text content_size = len(content) if content else 0 mime_type = getattr(content_item, "mimeType", "text/plain") elif hasattr(content_item, "blob"): # For binary content, we'll encode as base64 string import base64 content = base64.b64encode(content_item.blob).decode("utf-8") content_size = len(content_item.blob) mime_type = getattr(content_item, "mimeType", "application/octet-stream") else: content = str(content_item) content_size = len(content) # User-facing success update await ctx.info(f"Resource '{uri}' read successfully ({content_size} bytes)") # Detailed technical log logger.info( f"Resource '{uri}' read successfully", extra={ "resource_uri": uri, "content_size": content_size, "duration_ms": resource_elapsed_ms, }, ) return { "success": True, "resource": { "uri": uri, "mimeType": mime_type, "content": content, }, "metadata": { "content_size": content_size, "request_time_ms": round(total_elapsed_ms, 2), "server_url": state.server_url, "connection_statistics": state.statistics, }, } except ConnectionError as e: elapsed_ms = (time.perf_counter() - start_time) * 1000 # User-facing error update await ctx.error(f"Not connected when reading resource '{uri}': {str(e)}") # Detailed technical log logger.error( f"Not connected when reading resource '{uri}': {str(e)}", extra={"resource_uri": uri, "duration_ms": elapsed_ms}, ) return { "success": False, "error": { "error_type": "not_connected", "message": str(e), "details": {"resource_uri": uri}, "suggestion": "Use connect_to_server() to establish a connection first", }, "resource": None, "metadata": { "request_time_ms": round(elapsed_ms, 2), }, } except Exception as e: elapsed_ms = (time.perf_counter() - start_time) * 1000 # Determine error type based on exception message error_type = "execution_error" suggestion = "Check the resource URI and retry" error_msg = str(e).lower() if "not found" in error_msg or "unknown resource" in error_msg or "no resource" in error_msg: error_type = "resource_not_found" suggestion = f"Resource '{uri}' does not exist on the server. Use list_resources() to see available resources" # User-facing error update await ctx.error(f"Failed to read resource '{uri}': {str(e)}") # Detailed technical log logger.error( f"Failed to read resource '{uri}': {str(e)}", extra={ "resource_uri": uri, "error_type": error_type, "duration_ms": elapsed_ms, }, ) # Increment error counter ConnectionManager.increment_stat("errors") return { "success": False, "error": { "error_type": error_type, "message": f"Failed to read resource '{uri}': {str(e)}", "details": { "resource_uri": uri, "exception_type": type(e).__name__, }, "suggestion": suggestion, }, "resource": None, "metadata": { "request_time_ms": round(elapsed_ms, 2), }, } - Input schema defined via Annotated type hint for 'uri' parameter and comprehensive docstring describing input/output format, error types, and return structure. Output is dict[str, Any] with success, resource (uri, mimeType, content), metadata.
async def read_resource( uri: Annotated[str, "URI of the resource to read (e.g., 'config://settings')"], ctx: Context ) -> dict[str, Any]: """Read a specific resource from the connected MCP server. Reads a resource by URI and returns its content along with metadata. Returns: Dictionary with resource content including: - success: True if resource was read successfully - resource: Object with uri, mimeType, and content - metadata: Content size and request timing Raises: Returns error dict for various failure scenarios: - not_connected: No active connection - resource_not_found: Resource doesn't exist on server - execution_error: Resource read failed """ - node-wrapper/python-src/src/mcp_test_mcp/server.py:98-98 (registration)Import statement that loads the resources.py module, triggering the @mcp.tool decorator to register 'read_resource' (and list_resources) on the shared FastMCP 'mcp' instance.
from .tools import connection, tools, resources, prompts, llm