|
| 1 | +"""Clientside callbacks resource.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import json |
| 6 | +from typing import Any |
| 7 | + |
| 8 | +from mcp.types import ( |
| 9 | + ReadResourceResult, |
| 10 | + Resource, |
| 11 | + ResourceTemplate, |
| 12 | + TextResourceContents, |
| 13 | +) |
| 14 | + |
| 15 | +from dash import get_app |
| 16 | +from dash._utils import clean_property_name, split_callback_id |
| 17 | + |
| 18 | +URI = "dash://clientside-callbacks" |
| 19 | + |
| 20 | + |
| 21 | +def get_resource() -> Resource | None: |
| 22 | + if not _get_clientside_callbacks(): |
| 23 | + return None |
| 24 | + return Resource( |
| 25 | + uri=URI, |
| 26 | + name="dash_clientside_callbacks", |
| 27 | + description=( |
| 28 | + "Actions the user can take manually in the browser " |
| 29 | + "to affect clientside state. Inputs describe the " |
| 30 | + "components that can be changed to trigger an effect. " |
| 31 | + "Outputs describe the components that will change " |
| 32 | + "in response." |
| 33 | + ), |
| 34 | + mimeType="application/json", |
| 35 | + ) |
| 36 | + |
| 37 | + |
| 38 | +def get_template() -> ResourceTemplate | None: |
| 39 | + return None |
| 40 | + |
| 41 | + |
| 42 | +def read_resource(uri: str = "") -> ReadResourceResult: |
| 43 | + data = { |
| 44 | + "description": ( |
| 45 | + "These are actions that the user can take manually in the " |
| 46 | + "browser to affect the clientside state. Inputs describe " |
| 47 | + "the components that can be changed to trigger an effect. " |
| 48 | + "Outputs describe the components that will change in " |
| 49 | + "response to the effect." |
| 50 | + ), |
| 51 | + "callbacks": _get_clientside_callbacks(), |
| 52 | + } |
| 53 | + return ReadResourceResult( |
| 54 | + contents=[ |
| 55 | + TextResourceContents( |
| 56 | + uri=URI, |
| 57 | + mimeType="application/json", |
| 58 | + text=json.dumps(data, default=str), |
| 59 | + ) |
| 60 | + ] |
| 61 | + ) |
| 62 | + |
| 63 | + |
| 64 | +def _get_clientside_callbacks() -> list[dict[str, Any]]: |
| 65 | + """Get input/output mappings for clientside callbacks.""" |
| 66 | + app = get_app() |
| 67 | + callbacks = [] |
| 68 | + callback_map = getattr(app, "callback_map", {}) |
| 69 | + |
| 70 | + for output_id, callback_info in callback_map.items(): |
| 71 | + if "callback" in callback_info: |
| 72 | + continue |
| 73 | + normalize_deps = lambda deps: [ |
| 74 | + { |
| 75 | + "component_id": str(d.get("id", "unknown")), |
| 76 | + "property": d.get("property", "unknown"), |
| 77 | + } |
| 78 | + for d in deps |
| 79 | + ] |
| 80 | + parsed = split_callback_id(output_id) |
| 81 | + if isinstance(parsed, dict): |
| 82 | + parsed = [parsed] |
| 83 | + outputs = [ |
| 84 | + {"component_id": p["id"], "property": clean_property_name(p["property"])} |
| 85 | + for p in parsed |
| 86 | + ] |
| 87 | + callbacks.append( |
| 88 | + { |
| 89 | + "outputs": outputs, |
| 90 | + "inputs": normalize_deps(callback_info.get("inputs", [])), |
| 91 | + "state": normalize_deps(callback_info.get("state", [])), |
| 92 | + } |
| 93 | + ) |
| 94 | + |
| 95 | + return callbacks |
0 commit comments