Source code for tooluniverse.rnacentral_tool
import json
import re
from typing import Any, Dict
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from tooluniverse.tool_registry import register_tool
_EBI_SEARCH = "https://www.ebi.ac.uk/ebisearch/ws/rest/rnacentral"
_URS = re.compile(r"^(URS[0-9A-F]{10})(?:[_/](\d+))?$", re.I)
[docs]
def _http_get(
url: str, headers: Dict[str, str] | None = None, timeout: int = 30
) -> Dict[str, Any]:
req = Request(url, headers=headers or {})
with urlopen(req, timeout=timeout) as resp:
data = resp.read()
try:
return json.loads(data.decode("utf-8", errors="ignore"))
except Exception:
return {"raw": data.decode("utf-8", errors="ignore")}
[docs]
@register_tool(
"RNAcentralSearchTool",
config={
"name": "RNAcentral_search",
"type": "RNAcentralSearchTool",
"description": "Search ncRNA records by keyword (RNA or gene name) or fetch one by RNAcentral accession.",
"parameter": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "RNA or gene name (e.g. 'MALAT1', 'let-7') or an RNAcentral accession (URS0000D5CD72 or URS0000D5CD72_9606).",
},
"page_size": {
"type": "integer",
"default": 10,
"minimum": 1,
"maximum": 100,
},
"page": {
"type": "integer",
"default": 1,
"minimum": 1,
},
},
"required": ["query"],
},
"settings": {"base_url": "https://rnacentral.org/api/v1", "timeout": 60},
},
)
class RNAcentralSearchTool:
[docs]
def run(self, arguments: Dict[str, Any]):
base = self.tool_config.get("settings", {}).get(
"base_url", "https://rnacentral.org/api/v1"
)
timeout = int(self.tool_config.get("settings", {}).get("timeout", 30))
query = {
"query": (arguments.get("query") or "").strip(),
"page_size": int(arguments.get("page_size") or 10),
"page": max(int(arguments.get("page") or 1), 1),
}
if not query["query"]:
return {
"status": "error",
"error": "query is required (e.g. 'MALAT1', 'let-7', 'URS0000D5CD72').",
"source": "RNAcentral",
"success": False,
}
# The RNAcentral /rna endpoint has no text search: it ignores ``query`` and
# lists the same records for every value. Keyword searches therefore go
# through EBI Search's RNAcentral index; only an accession is fetched from
# the RNAcentral API itself.
accession = _URS.match(query["query"])
size = max(1, min(query["page_size"], 100))
start = (query["page"] - 1) * size
try:
if accession:
urs, taxid = accession.group(1).upper(), accession.group(2)
url = f"{base}/rna/{urs}" + (f"/{taxid}" if taxid else "")
record = _http_get(
url, headers={"Accept": "application/json"}, timeout=timeout
)
data = {"count": 1, "next": None, "previous": None, "results": [record]}
endpoint = "rna"
else:
params = {
"query": query["query"],
"format": "json",
"size": size,
"start": start,
"fields": "description,rna_type,length",
}
found = _http_get(f"{_EBI_SEARCH}?{urlencode(params)}", timeout=timeout)
results = []
for entry in found.get("entries", []):
fields = entry.get("fields", {})
rid = entry.get("id", "")
urs, _, taxid = rid.partition("_")
length = (fields.get("length") or [None])[0]
results.append(
{
"rnacentral_id": rid,
"url": f"{base}/rna/{urs}" + (f"/{taxid}" if taxid else ""),
"description": (fields.get("description") or [""])[0],
"rna_type": (fields.get("rna_type") or [""])[0],
"length": int(length) if length else None,
}
)
data = {
"count": found.get("hitCount", len(results)),
"next": None,
"previous": None,
"results": results,
}
endpoint = "ebisearch"
results = data["results"]
# EBI Search reports hitCount, so the page is the last one only
# when it reaches that total.
has_more = (
accession is None
and isinstance(data["count"], int)
and start + len(results) < data["count"]
)
return {
"status": "success",
"source": "RNAcentral",
"endpoint": endpoint,
"query": query,
"data": data,
"metadata": {
"returned": len(results),
"page": query["page"],
"page_size": query["page_size"],
"has_more": has_more,
"truncated": has_more,
"next_page": query["page"] + 1 if has_more else None,
},
"success": True,
}
except Exception as e:
return {
"status": "error",
"error": str(e),
"source": "RNAcentral",
"endpoint": "ebisearch" if not accession else "rna",
"success": False,
}
[docs]
@register_tool(
"RNAcentralGetTool",
config={
"name": "RNAcentral_get_by_accession",
"type": "RNAcentralGetTool",
"description": "Get RNAcentral entry by accession",
"parameter": {
"type": "object",
"properties": {
"accession": {"type": "string", "description": "RNAcentral accession"}
},
"required": ["accession"],
},
"settings": {"base_url": "https://rnacentral.org/api/v1", "timeout": 60},
},
)
class RNAcentralGetTool:
[docs]
def run(self, arguments: Dict[str, Any]):
base = self.tool_config.get("settings", {}).get(
"base_url", "https://rnacentral.org/api/v1"
)
timeout = int(self.tool_config.get("settings", {}).get("timeout", 30))
fields = self.tool_config.get("fields", {}) or {}
# 'region_overlap': when set, this tool instance queries the genomic
# overlap endpoint /overlap/region/{species}/{chr}:{start}-{end}, which
# takes coordinates rather than an accession. Reuses this class so no
# extra registration is needed.
if fields.get("region_overlap"):
return self._get_region_overlap(base, arguments, timeout)
acc = arguments.get("accession")
if not acc:
return {
"status": "error",
"error": "accession is required (e.g., 'URS000063A371').",
"source": "RNAcentral",
"success": False,
}
# 'sub_resources': list of rna/{accession}/{name} sub-endpoints to fetch
# and merge into one response (e.g. ['xrefs', 'publications']). This
# lets the same class serve the base record plus resolved
# cross-references and literature without adding a new @register_tool
# class. Falls back to the base rna/{accession} record when unset.
sub_resources = fields.get("sub_resources")
if sub_resources:
controls = self._sub_resource_controls(arguments, sub_resources)
if isinstance(controls, dict) and controls.get("status") == "error":
return controls
return self._get_sub_resources(base, acc, sub_resources, timeout, controls)
endpoint = "rna/{accession}"
url = f"{base}/rna/{acc}"
try:
data = _http_get(
url, headers={"Accept": "application/json"}, timeout=timeout
)
return {
"status": "success",
"source": "RNAcentral",
"endpoint": endpoint,
"accession": acc,
"data": data,
"success": True,
}
except Exception as e:
return {
"status": "error",
"error": str(e),
"source": "RNAcentral",
"endpoint": endpoint,
"accession": acc,
"success": False,
}
[docs]
@staticmethod
def _sub_resource_controls(arguments, sub_resources):
page_size = arguments.get("page_size", 10)
if (
isinstance(page_size, bool)
or not isinstance(page_size, int)
or not 1 <= page_size <= 50
):
return {
"status": "error",
"error": "page_size must be an integer between 1 and 50",
"parameter": "page_size",
"source": "RNAcentral",
"success": False,
}
controls = {"page_size": page_size}
for name in sub_resources:
parameter = f"{name}_page"
page = arguments.get(parameter, 1)
if isinstance(page, bool) or not isinstance(page, int) or page < 1:
return {
"status": "error",
"error": f"{parameter} must be a positive integer",
"parameter": parameter,
"source": "RNAcentral",
"success": False,
}
controls[parameter] = page
return controls
[docs]
def _get_sub_resources(self, base, acc, sub_resources, timeout, controls):
"""Fetch and merge one or more rna/{accession}/{name} sub-endpoints."""
data: Dict[str, Any] = {}
errors = []
for name in sub_resources:
page = controls[f"{name}_page"]
params = {
"format": "json",
"page_size": controls["page_size"],
"page": page,
}
url = f"{base}/rna/{acc}/{name}/?{urlencode(params)}"
try:
payload = _http_get(
url, headers={"Accept": "application/json"}, timeout=timeout
)
results = (
payload.get("results", []) if isinstance(payload, dict) else []
)
count = (
payload.get("count")
if isinstance(payload, dict) and payload.get("count") is not None
else len(results)
)
has_more = bool(payload.get("next"))
truncated = page > 1 or has_more
data[name] = {
"count": count,
"results": results,
"returned": len(results),
"page": page,
"page_size": controls["page_size"],
"has_more": has_more,
"next_page": page + 1 if has_more else None,
"previous_page": (page - 1 if payload.get("previous") else None),
"truncated": truncated,
"result_set_complete": not truncated,
}
except Exception as e:
errors.append(f"{name}: {e}")
# Only a hard failure (no sub-resource succeeded) returns an error.
if not data and errors:
return {
"status": "error",
"error": "; ".join(errors),
"source": "RNAcentral",
"endpoint": "rna/{accession}/[xrefs,publications]",
"accession": acc,
"success": False,
}
return {
"status": "success",
"source": "RNAcentral",
"endpoint": "rna/{accession}/[" + ",".join(sub_resources) + "]",
"accession": acc,
"data": data,
"partial_errors": errors or None,
"metadata": {
"publication_total_is_record_count": True,
"unique_pmids_not_guaranteed": "publications" in sub_resources,
"warning": (
"RNAcentral may repeat a PMID with different citation "
"metadata; publications.count is not guaranteed to be a "
"unique PMID count."
if "publications" in sub_resources
else None
),
},
"success": True,
}
[docs]
def _get_region_overlap(self, base, arguments, timeout):
"""Query /overlap/region/{species}/{chr}:{start}-{end} for ncRNAs.
Accepts either an explicit 'region' string ('2:39745816-39826679') or
the discrete 'chromosome', 'start', 'end' parameters. Returns the list
of overlapping RNAcentral transcripts and their exon features, split
into transcripts/exons for easy consumption.
"""
endpoint = "overlap/region/{species}/{region}"
species = (arguments.get("species") or "homo_sapiens").strip()
region = (arguments.get("region") or "").strip()
if not region:
chrom = arguments.get("chromosome")
start = arguments.get("start")
end = arguments.get("end")
if chrom in (None, "") or start in (None, "") or end in (None, ""):
return {
"status": "error",
"error": "Provide 'region' (e.g. '2:39745816-39826679') OR all of "
"'chromosome', 'start', 'end'.",
"source": "RNAcentral",
"endpoint": endpoint,
"success": False,
}
region = f"{chrom}:{start}-{end}"
url = f"{base}/overlap/region/{species}/{region}"
try:
data = _http_get(
url, headers={"Accept": "application/json"}, timeout=timeout
)
except Exception as e:
return {
"status": "error",
"error": str(e),
"source": "RNAcentral",
"endpoint": endpoint,
"region": region,
"success": False,
}
# The overlap endpoint returns a JSON array of feature dicts. Anything
# else (e.g. {"raw": ...}, {"error": ...}, {"message": ...}) is a
# server-side error or unexpected payload.
if not isinstance(data, list):
detail = ""
if isinstance(data, dict):
detail = data.get("error") or data.get("message") or data.get("raw")
return {
"status": "error",
"error": "RNAcentral overlap endpoint returned no feature list"
+ (f": {detail}" if detail else "."),
"source": "RNAcentral",
"endpoint": endpoint,
"region": region,
"success": False,
}
transcripts = [f for f in data if f.get("feature_type") == "transcript"]
exons = [f for f in data if f.get("feature_type") == "exon"]
return {
"status": "success",
"source": "RNAcentral",
"endpoint": endpoint,
"species": species,
"region": region,
"data": {
"features": data,
"transcripts": transcripts,
"exons": exons,
"transcript_count": len(transcripts),
"exon_count": len(exons),
},
"success": True,
}