retrieve_data_tool
Retrieve specific video segments and associated metadata from the Ragie index using a query. Returns text, document name, and timestamps for precise content extraction.
Instructions
Retrieves data from the Ragie index based on the query. The data is returned as a list of dictionaries, each containing the following keys:
- text: The text of the retrieved chunk
- document_name: The name of the document the chunk belongs to
- start_time: The start time of the chunk
- end_time: The end time of the chunk
Args:
query (str): The query to retrieve data from the Ragie index.
Returns:
list[dict]: The retrieved data.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Implementation Reference
- server.py:24-43 (handler)Handler function decorated with @mcp.tool() that implements the retrieve_data_tool by calling the retrieve_data helper function from main.py. Includes type hints and docstring serving as schema.@mcp.tool() def retrieve_data_tool(query: str) -> list[dict]: """ Retrieves data from the Ragie index based on the query. The data is returned as a list of dictionaries, each containing the following keys: - text: The text of the retrieved chunk - document_name: The name of the document the chunk belongs to - start_time: The start time of the chunk - end_time: The end time of the chunk Args: query (str): The query to retrieve data from the Ragie index. Returns: list[dict]: The retrieved data. """ try: content = retrieve_data(query) return content except Exception as e: return f"Failed to retrieve data: {str(e)}"
- main.py:86-109 (helper)Core helper function that performs the actual data retrieval using the Ragie client, formatting the results into list of dicts as expected by the tool.def retrieve_data(query): try: logger.info(f"Retrieving data for query: {query}") retrieval_response = ragie.retrievals.retrieve(request={ "query": query }) content = [ { **chunk.document_metadata, "text": chunk.text, "document_name": chunk.document_name, "start_time": chunk.metadata.get("start_time"), "end_time": chunk.metadata.get("end_time") } for chunk in retrieval_response.scored_chunks ] logger.info(f"Successfully retrieved {len(content)} chunks") return content except Exception as e: logger.error(f"Failed to retrieve data: {str(e)}") raise