Skip to content

Client Python Library

The Rhea client Python library allows you to interact with the Rhea MCP server similar to the FastMCP client library.

The client library utilizes both MCP and REST protocols of the Rhea server to allow for MCP tool calling and REST API file handling.

See examples on how to use the Client library.

RheaClient(hostname: str, port: int, secure: bool = False)

Bases: RheaMCPClientBase, RheaRESTClientBase

A client class to interact with both the Rhea Model Context Protocol (MCP) server and REST backend.

The client class provides similar high-level interface to the one found within the Python MCP SDK.

The class also provides additional utilities to interact with the REST backend such as file upload and downloads.

Example: 

async with RheaClient('localhost', 3001) as client:
    # MCP call
    tools = await client.find_tools('I need a tool to convert FASTA to FASTQ')
    # REST call
    key = await client.upload_file('test.txt')

Source code in rhea/client/__init__.py 
31
32
33
34
35
36
37
38
39
40
def __init__(self, hostname: str, port: int, secure: bool = False):
    self._hostname = hostname
    self._port = port
    self._secure = secure

    self._mcp_ctx: Optional[RheaMCPClient] = None
    self._mcp_client: Optional[RheaMCPClient] = None

    self._rest_ctx: Optional[RheaRESTClient] = None
    self._rest_client: Optional[RheaRESTClient] = None

list_tools() -> list[Tool] async

List the currently available tools on the server for this session.

Returns:

Type Description
list[Tool]

list[Tool]: A list of MCP tool definitions.
Source code in rhea/client/__init__.py 
68
69
70
71
async def list_tools(self) -> list[Tool]:
    if self._mcp_client is not None:
        return await self._mcp_client.list_tools()
    raise RuntimeError("`mcp_client` is None")

find_tools(query: str) -> list[dict] async

Find available tools on the MCP server that match the query.

This method searches for tools matching the provided query string and returns their descriptions.

Parameters:

Name Type Description Default
query str

The search query to find relevant tools.

 required

Returns:

Type Description
list[dict]

list[dict]: A list of tool descriptions matching the query.

Raises:

Type Description
RuntimeError

If the client session fails to initialize or used outside of a context manager.
Source code in rhea/client/__init__.py 
73
74
75
76
async def find_tools(self, query: str) -> list[dict]:
    if self._mcp_client is not None:
        return await self._mcp_client.find_tools(query)
    raise RuntimeError("`mcp_client` is None")

call_tool(name: str, arguments: dict[str, Any]) -> dict | None async

Call a specific tool on the MCP server with the given arguments.

Parameters:

Name Type Description Default
name str

The name of the tool to call.

 required
arguments dict[str, Any]

The arguments to pass to the tool.

 required

Returns:

Type Description
dict | None

dict | None: The structured content of the tool's response, or None if there is no structured content.

Raises:

Type Description
RuntimeError

If the client session fails to initialize or used outside of a context manager.
Source code in rhea/client/__init__.py 
78
79
80
81
async def call_tool(self, name: str, arguments: dict[str, Any]) -> dict | None:
    if self._mcp_client is not None:
        return await self._mcp_client.call_tool(name, arguments)
    raise RuntimeError("`mcp_client` is None")

list_resources() -> list[Resource] async

List all available resources from the Rhea MCP server.

This asynchronous method retrieves a list of all resources accessible through the initialized Rhea client. The client must have an active session before calling this method.

Returns:

Name Type Description
ListResourcesResult list[Resource]

A result object containing the list of available resources.

Raises: RuntimeError: If the client session fails to initialize or used outside of a context manager.

Source code in rhea/client/__init__.py 
83
84
85
86
async def list_resources(self) -> list[Resource]:
    if self._mcp_client is not None:
        return await self._mcp_client.list_resources()
    raise RuntimeError("`mcp_client` is None")

read_resource(uri: AnyUrl) -> list[TextResourceContents | BlobResourceContents] async

Read a specific resource from the Rhea MCP server by its URI.

This method retrieves the contents of a resource identified by the provided URI. The resource contents can be either text or binary data.

Parameters:

Name Type Description Default
uri AnyUrl

The URI of the resource to read.

 required

Returns:

Type Description
list[TextResourceContents | BlobResourceContents]

list[TextResourceContents | BlobResourceContents]: A list of resource contents, which can be either text or binary data.

Raises:

Type Description
RuntimeError

If the client session fails to initialize or used outside of a context manager.
Source code in rhea/client/__init__.py 
88
89
90
91
92
93
async def read_resource(
    self, uri: AnyUrl
) -> list[TextResourceContents | BlobResourceContents]:
    if self._mcp_client is not None:
        return await self._mcp_client.read_resource(uri)
    raise RuntimeError("`mcp_clien` is None")

upload_file(path: str, name: str | None = None, timeout: int = 300, chunk_size: int = 1 << 20) -> dict async

Upload a file from local directory to Rhea MCP server.

Parameters:

Name Type Description Default
path str

Path of local file to upload.

 required
name str

Optional filename to use on the server side. Defaults to source filename.

 None
timeout int

Request timeout in seconds. Defaults to 300 seconds.

 300
chunk_size int

Chunk size to read and upload file. Defaults to 1MB.

 1 << 20

Returns:

Name Type Description
dict dict

Server response

Raises:

Type Description
RuntimeError

If the client session fails to initialize or used outside of a context manager.
Source code in rhea/client/__init__.py 
 95
 96
 97
 98
 99
100
101
102
103
104
async def upload_file(
    self,
    path: str,
    name: str | None = None,
    timeout: int = 300,
    chunk_size: int = 1 << 20,
) -> dict:
    if self._rest_client is not None:
        return await self._rest_client.upload_file(path, name, timeout, chunk_size)
    raise RuntimeError("`rest_client` is None")

download_file(key: str, output_directory: Path = Path.cwd(), timeout: int = 300, chunk_size: int = 1 << 20) -> int async

Download a file to local directory from Rhea MCP server.

Parameters:

Name Type Description Default
key str

File key of desired file to download.

 required
output_directory Path

Output directory to write to. Defaults to current working directory.

 cwd()
timeout int

Request timeout in seconds. Defaults to 300 seconds.

 300
chunk_size int

Chunk size for download stream. Defaults to 1MB.

 1 << 20

Returns:

Name Type Description
int int

Size of downloaded file in bytes.

Raises:

Type Description
RuntimeError

If the client session fails to initialize or used outside of a context manager.
Source code in rhea/client/__init__.py 
106
107
108
109
110
111
112
113
114
115
116
117
async def download_file(
    self,
    key: str,
    output_directory: Path = Path.cwd(),
    timeout: int = 300,
    chunk_size: int = 1 << 20,
) -> int:
    if self._rest_client is not None:
        return await self._rest_client.download_file(
            key, output_directory, timeout, chunk_size
        )
    raise RuntimeError("`rest_client` is None")

metrics() -> dict[str, list[dict]] async

Get Prometheus metrics from Rhea MCP server.

See: Prometheus Specification

Returns:

Type Description
dict[str, list[dict]]

dict[str, list[dict]]: Server metrics

Raises:

Type Description
RuntimeError

If the client session fails to initialize or used outside of a context manager.
Source code in rhea/client/__init__.py 
119
120
121
122
async def metrics(self) -> dict[str, list[dict]]:
    if self._rest_client is not None:
        return await self._rest_client.metrics()
    raise RuntimeError("`rest_client` is None")