Spaces:

kalanosdev
/

kalanos-mcp

Running

Clarify read-only LeRobot search and quality report tools

8d2d208 verified 5 days ago

12.9 kB

"""HF-discoverable read-only adapter; no imports from the grading application."""
import asyncio
import json
from pathlib import Path

import gradio as gr
import httpx
from jsonschema import Draft7Validator
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

UPSTREAM = "https://mcp.kalanos.ai/mcp"
SCHEMAS = json.loads(Path(__file__).with_name("upstream-tools.json").read_text())
ALLOWED = frozenset({"list_datasets", "search_datasets", "get_report", "get_findings", "search_episodes", "search_combined_datasets"})
assert set(SCHEMAS) == ALLOWED


async def forward(name: str, arguments: dict) -> str:
    """Validate a fixed read-only tool and forward anonymously with a total deadline."""
    if name not in ALLOWED:
        raise ValueError("Tool is not allowed by this read-only adapter.")
    arguments = {key: value for key, value in arguments.items() if value is not None}
    errors = sorted(Draft7Validator(SCHEMAS[name]).iter_errors(arguments), key=lambda e: str(e.path))
    if errors:
        raise ValueError("Invalid tool arguments: " + errors[0].message)
    try:
        async with asyncio.timeout(30):
            # Fixed public origin; never forward HF tokens, cookies, proxy/netrc auth or client headers.
            async with httpx.AsyncClient(timeout=20, trust_env=False, follow_redirects=False) as client:
                async with streamable_http_client(UPSTREAM, http_client=client) as (read, write, _):
                    async with ClientSession(read, write) as session:
                        await session.initialize()
                        result = await session.call_tool(name, arguments)
                        envelope = result.model_dump(mode="json", by_alias=True, exclude_none=True)
    except (TimeoutError, ExceptionGroup, httpx.HTTPError) as exc:
        raise gr.Error("Public report service unavailable or timed out; retry later. No audit was started.") from exc
    if envelope.get("isError"):
        raise gr.Error("Upstream read-only error: " + json.dumps(envelope, ensure_ascii=False))
    return json.dumps(envelope, ensure_ascii=False)


async def list_datasets(query: str | None = None, after: str | None = None, limit: int = 10, format: str | None = None, robot: str | None = None, license: str | None = None, report_kind: str | None = None, min_readiness: float | None = None, min_episodes: int | None = None) -> str:
    """Read-only: Browse public LeRobot and robotics datasets already processed by Kalanos Hub alphabetically by slug, 10 per page by default (limit 1–25); never returns the whole catalog. Pass next_after back with the same arguments until it is null. Returns readiness, episode count, robot, format, license and whether Task/Objects/Scene labels exist. Optional literal substring query over slug/source_ref and exact filters (format, robot, license, report_kind, min_readiness, min_episodes). For keyword or relevance search use search_datasets. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
    return await forward("list_datasets", locals())


async def search_datasets(q: str | None = None, format: list[str] | None = None, robot: list[str] | None = None, license: list[str] | None = None, report_kind: list[str] | None = None, min_episodes: int | None = None, max_size_gb: float | None = None, min_hours: float | None = None, page: int = 1) -> str:
    """Read-only: Search public LeRobot and robotics datasets in Kalanos Hub by relevance: q is matched against the dataset slug, source ref and report summary text (robot names, tasks, formats, descriptions), ranked bm25; without q results are ranked by score. Filters: format[], robot[], license[], report_kind[] (OR within a list, AND across), min_episodes, max_size_gb, min_hours. 25 results per page with total and facet counts; follow page.next_page. Then call get_report with the returned report_id to inspect the existing data quality report. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
    return await forward("search_datasets", locals())


async def get_report(report_id: str | None = None, dataset: str | None = None, report_revision: str | None = None, dataset_revision: str | None = None) -> str:
    """Read-only: Retrieve the current existing public data quality report for a LeRobot or robotics dataset by report_id or dataset slug (e.g. lerobot/pusht): readiness, decision counts, coverage, attention queue, scores, dataset metadata, immutable report ETag, and Task/Objects/Scene label facets when published. Optional revision checks reject stale data; historical revisions are not retained. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
    return await forward("get_report", locals())


async def get_findings(report_revision: str, report_id: str | None = None, dataset: str | None = None, dataset_revision: str | None = None, after: int = 0, limit: int = 5) -> str:
    """Read-only: Retrieve recorded episode-level data quality findings and evidence for LeRobot and robotics datasets at an exact report revision, each with its Task/Objects/Scene labels when published. after is an episode offset, including episodes without findings. No episode selection or new evaluation. If output is too large lower limit. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
    return await forward("get_findings", locals())


async def search_episodes(q: str | None = None, dataset: str | None = None, task: list[str] | None = None, objects: list[str] | None = None, scene: list[str] | None = None, after: int = 0, limit: int = 20) -> str:
    """Read-only: Find public LeRobot and robotics episodes to review by Task family, object names and Scene within one dataset (source_ref, e.g. lerobot/pusht). Supply a nonempty dataset; unscoped calls return search_scope_required. Use search_combined_datasets for broad discovery. Filters AND across fields, OR within a field; q is a word match over the task description and labels. Returns episode rows with labels and review status plus facet counts in datasets. Labels are not grading results: call get_report / get_findings for quality. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
    return await forward("search_episodes", locals())


async def search_combined_datasets(q: str | None = None, dataset: str | None = None, revision: str | None = None, task: list[str] | None = None, objects: list[str] | None = None, scene: list[str] | None = None, format: list[str] | None = None, robot: list[str] | None = None, license: list[str] | None = None, report_kind: list[str] | None = None, min_episodes: int | None = None, max_size_gb: float | None = None, min_hours: float | None = None, page: int = 1, per_page: int = 20, direction: str = "desc", facets: bool = False) -> str:
    """Read-only: Discover public LeRobot and robotics datasets across Kalanos Hub using task/objects/scene labels together with dataset metadata filters. Returns dataset pages, not globally paginated individual episodes. Set facets=True to request full-result facets/counts; otherwise totals can be null and use page.next_page/has_more. Same-episode AND across episode filters, OR within groups. Without episode facets text may match dataset metadata or labels; with episode facets it must match the same episode. Sort by readiness (null last); direction asc/desc; per_page 20/40/80; page is clamped by the Hub. Use source_ref with scoped search_episodes for episode details. Labels are not quality findings. Never grades or publishes. Returns the upstream result envelope as JSON text."""
    return await forward("search_combined_datasets", locals())


with gr.Blocks(title="Kalanos LeRobot & Robotics Dataset Search MCP") as demo:
    gr.Markdown("""# Kalanos LeRobot & Robotics Dataset Search MCP
Search LeRobot and robotics datasets, inspect existing data quality reports, and find episodes to review in your AI assistant.
## What you can do
- **Find relevant robotics datasets.** Search by task, objects, scene, robot, format, license, episode count and dataset size, using available metadata and labels.
- **Inspect LeRobot quality reports.** See recorded decisions, assessment coverage and limitations before choosing what to review.
- **Trace findings to episodes.** Explore recorded issues and evidence, then search within a dataset for published task, object and scene labels.
- **Compare assessments with context.** Ask your assistant to compare report dates, revisions and measured scope instead of treating scores as interchangeable.
This is read-only access to existing public reports. It does not run new grading, download dataset shards or publish reports. Labels and scores do not establish task success, hardware health or training performance.
## Setup
### Connect through Hugging Face
1. Open the app tab and use the MCP badge to add this Space to your MCP tools.
2. In [Hugging Face MCP settings](https://huggingface.co/settings/mcp), choose your client and follow its generated setup instructions. Hugging Face requires a token with read access for this route.
3. Refresh or restart your client and confirm the Kalanos tools are available.
See the [Hugging Face setup guide](https://huggingface.co/docs/hub/en/spaces-mcp-servers).
### Connect a remote MCP client directly
Use the public endpoint **https://mcp.kalanos.ai/mcp**, transport **Streamable HTTP**, and **no authentication**. No API key is needed for the direct public endpoint. Enable the server and refresh its tools.
For the adapter hosted in this Space, open **Use via API → MCP** in the app footer and copy the connection URL shown there.
[Client-specific setup and usage](https://kalanos.ai/mcp) covers Claude, ChatGPT and other MCP clients. Availability depends on your account and workspace settings.
## Sample questions
- "Find robotics datasets for pick-and-place. Show robot type, episode count and available quality reports. Explain missing metadata."
- "Search for lerobot/pusht, then use the returned report ID to show its current assessment. What was measured, what was not measured, and what needs review?"
- "Find the reports for lerobot/pusht and lerobot/aloha_sim_insertion_human. Compare their recorded findings and coverage. Explain whether the assessments are comparable."
- "For lerobot/berkeley_autolab_ur5, fetch the current report and its recorded episode findings. Which episodes should I inspect first, and what evidence supports that?"
- "Within [dataset ID], find episodes matching [task or object]. Distinguish published labels from quality findings, and explain any missing label coverage."
For reliable report lookup, search or list first and use the returned report_id. Fetch the current report_revision before requesting findings. Results are paginated; missing evidence is not a pass.
## Available tools
Six read-only tools retrieve existing public data and reports:
- list_datasets: browse public LeRobot and robotics datasets already processed by Kalanos, with metadata and report filters.
- search_datasets: find LeRobot and robotics datasets by keyword, ranked relevance and dataset filters.
- search_combined_datasets: discover datasets using metadata together with published task, object and scene labels; returns datasets, not a global episode list.
- get_report: retrieve an existing data quality report with recorded decisions, scores, assessment coverage and limitations. Use the report_id returned by search or list.
- get_findings: retrieve recorded episode-level quality findings and evidence at the exact report_revision returned by get_report.
- search_episodes: find episodes to review within one dataset using published task, object and scene labels. Labels are separate from quality findings.
Inspect the [tool schema](/gradio_api/mcp/schema). Anonymous rate limits apply to this shared Space.
[Browse reports](https://hub.kalanos.ai/datasets) · [Request a new assessment](https://kalanos.ai/submit-dataset)
""")
    for fn in (list_datasets, search_datasets, get_report, get_findings, search_episodes, search_combined_datasets):
        gr.api(fn, api_name=fn.__name__, concurrency_limit=4, concurrency_id="public_reads")

if __name__ == "__main__":
    demo.queue(max_size=16).launch(mcp_server=True)

	"""HF-discoverable read-only adapter; no imports from the grading application."""
	import asyncio
	import json
	from pathlib import Path

	import gradio as gr
	import httpx
	from jsonschema import Draft7Validator
	from mcp import ClientSession
	from mcp.client.streamable_http import streamable_http_client

	UPSTREAM = "https://mcp.kalanos.ai/mcp"
	SCHEMAS = json.loads(Path(__file__).with_name("upstream-tools.json").read_text())
	ALLOWED = frozenset({"list_datasets", "search_datasets", "get_report", "get_findings", "search_episodes", "search_combined_datasets"})
	assert set(SCHEMAS) == ALLOWED


	async def forward(name: str, arguments: dict) -> str:
	"""Validate a fixed read-only tool and forward anonymously with a total deadline."""
	if name not in ALLOWED:
	raise ValueError("Tool is not allowed by this read-only adapter.")
	arguments = {key: value for key, value in arguments.items() if value is not None}
	errors = sorted(Draft7Validator(SCHEMAS[name]).iter_errors(arguments), key=lambda e: str(e.path))
	if errors:
	raise ValueError("Invalid tool arguments: " + errors[0].message)
	try:
	async with asyncio.timeout(30):
	# Fixed public origin; never forward HF tokens, cookies, proxy/netrc auth or client headers.
	async with httpx.AsyncClient(timeout=20, trust_env=False, follow_redirects=False) as client:
	async with streamable_http_client(UPSTREAM, http_client=client) as (read, write, _):
	async with ClientSession(read, write) as session:
	await session.initialize()
	result = await session.call_tool(name, arguments)
	envelope = result.model_dump(mode="json", by_alias=True, exclude_none=True)
	except (TimeoutError, ExceptionGroup, httpx.HTTPError) as exc:
	raise gr.Error("Public report service unavailable or timed out; retry later. No audit was started.") from exc
	if envelope.get("isError"):
	raise gr.Error("Upstream read-only error: " + json.dumps(envelope, ensure_ascii=False))
	return json.dumps(envelope, ensure_ascii=False)


	async def list_datasets(query: str \| None = None, after: str \| None = None, limit: int = 10, format: str \| None = None, robot: str \| None = None, license: str \| None = None, report_kind: str \| None = None, min_readiness: float \| None = None, min_episodes: int \| None = None) -> str:
	"""Read-only: Browse public LeRobot and robotics datasets already processed by Kalanos Hub alphabetically by slug, 10 per page by default (limit 1–25); never returns the whole catalog. Pass next_after back with the same arguments until it is null. Returns readiness, episode count, robot, format, license and whether Task/Objects/Scene labels exist. Optional literal substring query over slug/source_ref and exact filters (format, robot, license, report_kind, min_readiness, min_episodes). For keyword or relevance search use search_datasets. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
	return await forward("list_datasets", locals())


	async def search_datasets(q: str \| None = None, format: list[str] \| None = None, robot: list[str] \| None = None, license: list[str] \| None = None, report_kind: list[str] \| None = None, min_episodes: int \| None = None, max_size_gb: float \| None = None, min_hours: float \| None = None, page: int = 1) -> str:
	"""Read-only: Search public LeRobot and robotics datasets in Kalanos Hub by relevance: q is matched against the dataset slug, source ref and report summary text (robot names, tasks, formats, descriptions), ranked bm25; without q results are ranked by score. Filters: format[], robot[], license[], report_kind[] (OR within a list, AND across), min_episodes, max_size_gb, min_hours. 25 results per page with total and facet counts; follow page.next_page. Then call get_report with the returned report_id to inspect the existing data quality report. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
	return await forward("search_datasets", locals())


	async def get_report(report_id: str \| None = None, dataset: str \| None = None, report_revision: str \| None = None, dataset_revision: str \| None = None) -> str:
	"""Read-only: Retrieve the current existing public data quality report for a LeRobot or robotics dataset by report_id or dataset slug (e.g. lerobot/pusht): readiness, decision counts, coverage, attention queue, scores, dataset metadata, immutable report ETag, and Task/Objects/Scene label facets when published. Optional revision checks reject stale data; historical revisions are not retained. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
	return await forward("get_report", locals())


	async def get_findings(report_revision: str, report_id: str \| None = None, dataset: str \| None = None, dataset_revision: str \| None = None, after: int = 0, limit: int = 5) -> str:
	"""Read-only: Retrieve recorded episode-level data quality findings and evidence for LeRobot and robotics datasets at an exact report revision, each with its Task/Objects/Scene labels when published. after is an episode offset, including episodes without findings. No episode selection or new evaluation. If output is too large lower limit. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
	return await forward("get_findings", locals())


	async def search_episodes(q: str \| None = None, dataset: str \| None = None, task: list[str] \| None = None, objects: list[str] \| None = None, scene: list[str] \| None = None, after: int = 0, limit: int = 20) -> str:
	"""Read-only: Find public LeRobot and robotics episodes to review by Task family, object names and Scene within one dataset (source_ref, e.g. lerobot/pusht). Supply a nonempty dataset; unscoped calls return search_scope_required. Use search_combined_datasets for broad discovery. Filters AND across fields, OR within a field; q is a word match over the task description and labels. Returns episode rows with labels and review status plus facet counts in datasets. Labels are not grading results: call get_report / get_findings for quality. Never grades, downloads datasets, or publishes. Returns the upstream MCP result envelope as JSON text; upstream errors become tool errors. Use report_id returned by search/list for report lookup."""
	return await forward("search_episodes", locals())


	async def search_combined_datasets(q: str \| None = None, dataset: str \| None = None, revision: str \| None = None, task: list[str] \| None = None, objects: list[str] \| None = None, scene: list[str] \| None = None, format: list[str] \| None = None, robot: list[str] \| None = None, license: list[str] \| None = None, report_kind: list[str] \| None = None, min_episodes: int \| None = None, max_size_gb: float \| None = None, min_hours: float \| None = None, page: int = 1, per_page: int = 20, direction: str = "desc", facets: bool = False) -> str:
	"""Read-only: Discover public LeRobot and robotics datasets across Kalanos Hub using task/objects/scene labels together with dataset metadata filters. Returns dataset pages, not globally paginated individual episodes. Set facets=True to request full-result facets/counts; otherwise totals can be null and use page.next_page/has_more. Same-episode AND across episode filters, OR within groups. Without episode facets text may match dataset metadata or labels; with episode facets it must match the same episode. Sort by readiness (null last); direction asc/desc; per_page 20/40/80; page is clamped by the Hub. Use source_ref with scoped search_episodes for episode details. Labels are not quality findings. Never grades or publishes. Returns the upstream result envelope as JSON text."""
	return await forward("search_combined_datasets", locals())


	with gr.Blocks(title="Kalanos LeRobot & Robotics Dataset Search MCP") as demo:
	gr.Markdown("""# Kalanos LeRobot & Robotics Dataset Search MCP

	Search LeRobot and robotics datasets, inspect existing data quality reports, and find episodes to review in your AI assistant.

	## What you can do

	- Find relevant robotics datasets. Search by task, objects, scene, robot, format, license, episode count and dataset size, using available metadata and labels.
	- Inspect LeRobot quality reports. See recorded decisions, assessment coverage and limitations before choosing what to review.
	- Trace findings to episodes. Explore recorded issues and evidence, then search within a dataset for published task, object and scene labels.
	- Compare assessments with context. Ask your assistant to compare report dates, revisions and measured scope instead of treating scores as interchangeable.

	This is read-only access to existing public reports. It does not run new grading, download dataset shards or publish reports. Labels and scores do not establish task success, hardware health or training performance.

	## Setup

	### Connect through Hugging Face

	1. Open the app tab and use the MCP badge to add this Space to your MCP tools.
	2. In [Hugging Face MCP settings](https://huggingface.co/settings/mcp), choose your client and follow its generated setup instructions. Hugging Face requires a token with read access for this route.
	3. Refresh or restart your client and confirm the Kalanos tools are available.

	See the [Hugging Face setup guide](https://huggingface.co/docs/hub/en/spaces-mcp-servers).

	### Connect a remote MCP client directly

	Use the public endpoint https://mcp.kalanos.ai/mcp, transport Streamable HTTP, and no authentication. No API key is needed for the direct public endpoint. Enable the server and refresh its tools.

	For the adapter hosted in this Space, open Use via API → MCP in the app footer and copy the connection URL shown there.

	[Client-specific setup and usage](https://kalanos.ai/mcp) covers Claude, ChatGPT and other MCP clients. Availability depends on your account and workspace settings.

	## Sample questions

	- "Find robotics datasets for pick-and-place. Show robot type, episode count and available quality reports. Explain missing metadata."
	- "Search for lerobot/pusht, then use the returned report ID to show its current assessment. What was measured, what was not measured, and what needs review?"
	- "Find the reports for lerobot/pusht and lerobot/aloha_sim_insertion_human. Compare their recorded findings and coverage. Explain whether the assessments are comparable."
	- "For lerobot/berkeley_autolab_ur5, fetch the current report and its recorded episode findings. Which episodes should I inspect first, and what evidence supports that?"
	- "Within [dataset ID], find episodes matching [task or object]. Distinguish published labels from quality findings, and explain any missing label coverage."

	For reliable report lookup, search or list first and use the returned report_id. Fetch the current report_revision before requesting findings. Results are paginated; missing evidence is not a pass.


	## Available tools

	Six read-only tools retrieve existing public data and reports:

	- list_datasets: browse public LeRobot and robotics datasets already processed by Kalanos, with metadata and report filters.
	- search_datasets: find LeRobot and robotics datasets by keyword, ranked relevance and dataset filters.
	- search_combined_datasets: discover datasets using metadata together with published task, object and scene labels; returns datasets, not a global episode list.
	- get_report: retrieve an existing data quality report with recorded decisions, scores, assessment coverage and limitations. Use the report_id returned by search or list.
	- get_findings: retrieve recorded episode-level quality findings and evidence at the exact report_revision returned by get_report.
	- search_episodes: find episodes to review within one dataset using published task, object and scene labels. Labels are separate from quality findings.

	Inspect the [tool schema](/gradio_api/mcp/schema). Anonymous rate limits apply to this shared Space.

	[Browse reports](https://hub.kalanos.ai/datasets) · [Request a new assessment](https://kalanos.ai/submit-dataset)
	""")
	for fn in (list_datasets, search_datasets, get_report, get_findings, search_episodes, search_combined_datasets):
	gr.api(fn, api_name=fn.__name__, concurrency_limit=4, concurrency_id="public_reads")

	if __name__ == "__main__":
	demo.queue(max_size=16).launch(mcp_server=True)