berlin_analyze_datasets
Read-onlyIdempotent
Analyzes Berlin open datasets for relevance, freshness, and available formats to identify suitable data for your projects.
Instructions
Analysiert Datensaetze umfassend: Relevanz, Aktualitaet und verfuegbare Formate.
Kombiniert Suche mit Analyse der Metadaten und Ressourcen-Formate.
Besonders nuetzlich um herauszufinden, welche Daten verfuegbar sind
und wie aktuell sie sind.
Hinweis: Berlins CKAN hat keinen DataStore – Daten muessen ueber
die Ressourcen-URLs heruntergeladen werden.
Returns:
Umfassender Analyse-Report mit Relevanz, Aktualitaet und FormatenInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- Handler function for the 'berlin_analyze_datasets' tool.
@mcp.tool( name="berlin_analyze_datasets", annotations={ "title": "Datensaetze analysieren", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, ) async def berlin_analyze_datasets(params: AnalyzeDatasetInput) -> str: """Analysiert Datensaetze umfassend: Relevanz, Aktualitaet und verfuegbare Formate. Kombiniert Suche mit Analyse der Metadaten und Ressourcen-Formate. Besonders nuetzlich um herauszufinden, welche Daten verfuegbar sind und wie aktuell sie sind. Hinweis: Berlins CKAN hat keinen DataStore – Daten muessen ueber die Ressourcen-URLs heruntergeladen werden. Returns: Umfassender Analyse-Report mit Relevanz, Aktualitaet und Formaten """ try: result = await ckan_request( "package_search", { "q": params.query, "rows": params.max_datasets, "sort": "score desc", }, ) datasets = result["results"] total = result["count"] if not datasets: return f"Keine Datensaetze gefunden fuer '{params.query}'." lines = [ f"## Analyse: '{params.query}'", f"**{total} Datensaetze gefunden**, Top {len(datasets)} analysiert:\n", ] for i, ds in enumerate(datasets, 1): name = ds.get("name", "") title = ds.get("title", "?") modified = ds.get("metadata_modified", "?")[:10] resources = ds.get("resources", []) formats = sorted(set(r.get("format", "?") for r in resources)) # Berlin-specific extras extras = {e["key"]: e["value"] for e in ds.get("extras", [])} date_updated = extras.get("date_updated", "") geo_coverage = extras.get("geographical_coverage", "") lines.append(f"### {i}. {title}") lines.append(f"- **ID**: `{name}`") lines.append(f"- **Formate**: {', '.join(formats)}") lines.append(f"- **Ressourcen**: {len(resources)}") if params.include_freshness: lines.append(f"- **Letzte Aenderung**: {modified}") if date_updated: lines.append(f"- **Daten aktualisiert**: {date_updated}") if params.include_structure: for res in resources: res_format = res.get("format", "?") res_name = res.get("name", "Unbenannt") res_url = res.get("url", "") lines.append(f" - {res_name} ({res_format}): {res_url}") if geo_coverage: lines.append(f"- **Raeumliche Abdeckung**: {geo_coverage}") lines.append(f"- **URL**: {PORTAL_URL}/datensaetze/{name}\n") return "\n".join(lines) except Exception as e: return handle_api_error(e, "Datensatz-Analyse") - Input validation schema for the 'berlin_analyze_datasets' tool.
class AnalyzeDatasetInput(BaseModel): """Input fuer Datensatz-Analyse.""" model_config = ConfigDict(str_strip_whitespace=True, extra="forbid") query: str = Field( ..., description="Suchbegriff fuer die Analyse, z.B. 'Einwohner', 'Verkehr', 'Wohnen'", min_length=1, ) max_datasets: int = Field(default=5, description="Maximale Anzahl zu analysierender Datensaetze", ge=1, le=20) include_structure: bool = Field(default=True, description="Ressourcen-Formate einschliessen") include_freshness: bool = Field(default=True, description="Aktualitaets-Analyse einschliessen")