check_downloads
Retrieve the weekly download count for any npm or PyPI package by providing its name and registry.
Instructions
Check the weekly download count of a package on npm or PyPI.
Args: package_name: The package name — e.g. "react", "express", "flask", "requests" registry: "npm" or "pypi" (defaults to "npm")
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| package_name | Yes | ||
| registry | No | npm |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/grounded_mcp/server.py:70-78 (handler)The actual tool handler function for 'check_downloads'. It calls _query_fact with registry, package_name, and field 'weekly_downloads'.
@mcp.tool() async def check_downloads(package_name: str, registry: str = "npm") -> str: """Check the weekly download count of a package on npm or PyPI. Args: package_name: The package name — e.g. "react", "express", "flask", "requests" registry: "npm" or "pypi" (defaults to "npm") """ return await _query_fact(registry, package_name, "weekly_downloads") - src/grounded_mcp/server.py:70-70 (registration)The @mcp.tool() decorator registers 'check_downloads' as an MCP tool on the FastMCP server instance.
@mcp.tool() - src/grounded_mcp/server.py:18-44 (helper)The _query_fact helper function that check_downloads delegates to. It makes an HTTP GET request to the Grounded API and formats the response.
async def _query_fact(source: str, entity: str, field: str) -> str: """Query the Grounded API and return a formatted result.""" headers = {} if API_KEY: headers["X-API-Key"] = API_KEY async with httpx.AsyncClient() as client: resp = await client.get( f"{API_BASE}/v1/fact", params={"source": source, "entity": entity, "field": field}, headers=headers, timeout=10.0, ) if resp.status_code == 200: data = resp.json() return ( f"Value: {data['value']}\n" f"Source: {data['source_url']}\n" f"Fetched at: {data['fetched_at']}\n" f"Hash: {data['raw_response_hash']}\n" f"Tier: {data['tier']} (TTL: {data['ttl_seconds']}s)" ) elif resp.status_code == 404: detail = resp.json().get("detail", "Not found") return f"Not found: {detail}" else: return f"Error: HTTP {resp.status_code}"