Berlin Open Data MCP Server

Instructions

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

Implementation Reference

• src/berlin_opendata_mcp/server.py:313-394 (handler)

  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")

• src/berlin_opendata_mcp/server.py:298-311 (schema)

  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")