Examples
Rhea is a MCP-compliant server that can be used across a variety of LLM clients and libraries to perform dynamic discovery and execution of thousands of tools. While all MCP functionality can be performed with any MCP-compatible client libary, we provide a client libary to provide a simple interface with Rhea's tool workflow and file management.
The client can be installed from PyPI using the uv package manager:
Or with pip:
Tool Calling
Tool calling with Rhea is performed exclusively over the MCP protocol to provide universal support across many LLM clients. The Rhea client library provides several helper functions on-top of the FastMCP client libary for ease of use.
Finding Tools
A major feature of Rhea is the ability to dynamically propogate the tools made available on the server via Retrieval Augmented Generation (RAG). This is achieved by providing a natural language query to the find_tools() MCP tool which will perform a server-side RAG and populate the tools for this client session.
Using Rhea Client Library
The following script accepts a natural language query, such as "I need a tool to convert FASTA to FASTQ", and returns the relevant tools for this query.
This script can be found in examples/find_tools.py
# find_tools.py
import asyncio
from rhea.client import RheaClient
from argparse import ArgumentParser
from urllib.parse import urlparse
from mcp.types import Tool
parser = ArgumentParser(
description="Find relevant tools based on natural language query."
)
parser.add_argument("query", help="Natural language query")
parser.add_argument("--url", help="URL of MCP server", default="http://localhost:3001")
args = parser.parse_args()
async def main():
parsed_url = urlparse(args.url)
protocol = parsed_url.scheme
host = parsed_url.hostname
port = parsed_url.port
secure = protocol == "https"
async with RheaClient(host, port, secure) as client: # (1)!
await client.find_tools(args.query) # (2)!
tools: list[Tool] = await client.list_tools() # (3)!
print(tools)
if __name__ == "__main__":
asyncio.run(main()) # (4)!
- Open a connection with the MCP server with an asynchronous context manager.
- Call
find_tools()with a natural language query string. - List the tools avaiable on the server for this client.
- Make sure to run as a coroutine.
Using FastMCP
import asyncio
from argparse import ArgumentParser
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from mcp.types import Tool
parser = ArgumentParser(
description="Find relevant tools based on natural language query."
)
parser.add_argument("query", help="Natural language query")
parser.add_argument("--url", help="URL of MCP server", default="http://localhost:3001")
args = parser.parse_args()
async def main():
async with streamablehttp_client(f"{args.url}/mcp") as (
read,
write,
get_session_id_callback,
):
async with ClientSession(read, write) as session:
await session.initialize()
await http_client_session.call_tool(
"find_tools", {"query": "I need a tool to convert FASTA to FASTQ"}
)
tools: list[Tool] = await client.list_tools()
print(tools)
if __name__ == "__main__":
asyncio.run(main())
File Management
For many tools, you will need to manage input and output files for your tool calls. To assist with file management, the Rhea server exposes a REST API that allows for file uploads/downloads to the backend server.
Files stored within Rhea are keyed as UUIDv4 strings and are accepted as input for tool calls requiring input files.
Uploading a File
The following script lets you upload a file from your local directory to the Rhea MCP server.
The script will output a UUIDv4 string representing the file string.
This script can be found in examples/upload_file.py
# upload_file.py
import asyncio
from rhea.client import RheaClient
from argparse import ArgumentParser
from urllib.parse import urlparse
parser = ArgumentParser(description="Upload files to Rhea MCP server")
parser.add_argument("input_file", help="Input file")
parser.add_argument("--url", help="URL of MCP server", default="http://localhost:3001")
parser.add_argument("--name", required=False, help="Name for uploaded file.")
args = parser.parse_args()
async def main():
parsed_url = urlparse(args.url)
protocol = parsed_url.scheme
host = parsed_url.hostname
port = parsed_url.port
secure = protocol == "https"
async with RheaClient(host, port, secure) as client: # (1)!
result: dict = await client.upload_file(args.input_file, args.name) # (2)!
print(result)
print(result["key"]) # (3)!
if __name__ == "__main__":
asyncio.run(main()) # (4)!
- Open a connection with the MCP server with an asynchronous context manager.
- Call
upload_file()with the path of your file. - This is the file key for your uploaded file to be used as input for tools.
- Make sure to run as a coroutine.
Usage:
Downloading a File
The following scripts lets you download a file from the Rhea MCP server to a local directory.
This script can be found in examples/download_file.py
# download_file.py
import asyncio
from rhea.client import RheaClient
from argparse import ArgumentParser
from pathlib import Path
from urllib.parse import urlparse
parser = ArgumentParser(description="Download files from Rhea MCP server")
parser.add_argument("key", help="Key of desired file")
parser.add_argument("--url", help="URL of MCP server", default="http://localhost:3001")
parser.add_argument("--output-directory", help="Output directory", default=Path.cwd())
parser.add_argument("--output-name", help="Name of output file")
args = parser.parse_args()
async def main():
parsed_url = urlparse(args.url)
protocol = parsed_url.scheme
host = parsed_url.hostname
port = parsed_url.port
secure = protocol == "https"
async with RheaClient(host, port, secure) as client: # (1)!
await client.download_file(args.key, args.output_directory) # (2)!
if __name__ == "__main__":
asyncio.run(main()) # (3)!
- Open a connection with the MCP server with an asynchronous context manager.
- Call
download_file()with your desired file key and output path. - Make sure to run as a coroutine.
Usage: