Function _derive_output_ids

dash/mcp/primitives/tools/callback_adapter.py:433–479

Derive concrete output IDs from the resolved input entries. Extracts the wildcard key values from the LLM-provided concrete input IDs and substitutes them into the output pattern.

(
    output_pattern: WildcardId,
    resolved_inputs: list[CallbackInputs],
)

def _derive_output_ids(
    output_pattern: WildcardId,
    resolved_inputs: list[CallbackInputs],
) -> list[WildcardId] | None:
    """Derive concrete output IDs from the resolved input entries.

    Extracts the wildcard key values from the LLM-provided concrete
    input IDs and substitutes them into the output pattern.
    """
    wildcard_keys = [
        k
        for k, v in output_pattern.items()
        if isinstance(v, list) and len(v) == 1 and v[0] in _WILDCARD_VALUES
    ]
    if not wildcard_keys:
        return None

    def _substitute(item_id: WildcardId) -> WildcardId | None:
        if not isinstance(item_id, dict):
            return None
        output_id = dict(output_pattern)
        for wk in wildcard_keys:
            if wk in item_id:
                output_id[wk] = item_id[wk]
        return output_id

    for entry in resolved_inputs:
        # ALL/ALLSMALLER: nested array of {id, property, value} dicts
        if isinstance(entry, list) and entry:
            concrete_ids = []
            for item in entry:
                item_id = item.get("id")
                if isinstance(item_id, dict):
                    out = _substitute(item_id)
                    if out:
                        concrete_ids.append(out)
            if concrete_ids:
                return concrete_ids
        # MATCH: single {id, property, value} dict
        elif isinstance(entry, dict):
            entry_id = entry.get("id")
            if isinstance(entry_id, dict):
                out = _substitute(entry_id)
                if out:
                    return [out]

    return None