File size: 9,934 Bytes
e488ce7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 | """The agent surface: the same verbs as the CLI, typed.
<mcp-venv>/bin/python -m concept_lab.mcp
Built on the same MCPServer/pydantic shape the h3-audio-lab server uses, so
one venv serves both and the two read alike. The mcp and pydantic imports
are inside this module rather than the package, so a ComfyUI install with
neither still loads concept_lab fine; only running the server needs them.
What is deliberately NOT on this surface:
space_export with include_private. Export is a publishing act, packs can
encode a likeness, and a model cannot prove who is asking. The agent can
export the shareable set and can see exactly which packs were held back
and why; releasing a private one is a human's keystroke through the CLI.
Any verb that would run a model. Capture costs GPU time on a shared box
with production instances on it, so the agent plans and a human launches.
capture_plan gives the agent the whole cost picture for free.
An agent that can measure everything and publish nothing is the right shape
for this subsystem.
"""
from __future__ import annotations
import sys
from typing import Optional
from pydantic import BaseModel, Field
from mcp.server import MCPServer
from concept_lab import api
mcp = MCPServer(
name="concept-lab",
title="Concept Lab",
instructions=(
"Functional conditioning: measure what a reference does to the "
"model, factor it, compile it back. Records here are EVIDENCE and "
"carry their scope: a capture belongs to one model stack and a "
"factor belongs to the split it was fitted on. Nothing on this "
"surface runs a model or publishes a concept pack; plan the work "
"and hand the launch to a human. Imported records keep their origin "
"space id and are never your own measurements."),
)
# ------------------------------------------------------------- outputs
class SpaceOut(BaseModel):
space_id: str
label: str = ""
index_root: str = ""
tensor_root: str = ""
counts: dict = Field(default_factory=dict)
imported_spaces: list = Field(default_factory=list)
tensor_bytes: int = 0
class CorpusOut(BaseModel):
name: str
corpus_id: str
n_rows: int
by_split: dict = Field(default_factory=dict)
label_keys: list = Field(default_factory=list)
matched_groups: dict = Field(default_factory=dict)
singleton_groups: list = Field(default_factory=list)
containment: dict = Field(
default_factory=dict,
description="Sources appearing in more than one split, by "
"content_hash and by source_id. Any entry here means a "
"held-out score would be an extraction score.")
warnings: list = Field(
default_factory=list,
description="Protocol gaps. Not errors: a corpus missing its "
"adversarial split can still be captured, it just "
"cannot support a factor claim afterwards.")
class ConfoundOut(BaseModel):
target: str
n_rows: int
labels: dict = Field(
description="Per label: verdict perfect|partial|free. 'perfect' "
"means the two labels are in bijection over these rows "
"and no method separates them; the fix is another "
"example, not another solver.")
error: Optional[str] = None
class PlanOut(BaseModel):
corpus: str
backend: str
arms: list
forwards: int = Field(description="model forward passes this would cost")
tap_points: int = Field(description="tensors this would write")
tap: dict
cost: dict = Field(
default_factory=dict,
description="Forward count is not the cost unit on H3: a reference "
"block roughly doubles the packed sequence and per-step "
"cost is superlinear in it. Price by tokens x steps.")
process_launch_id: str = ""
protocol_notes: list = Field(
default_factory=list,
description="Constraints on how the capture may be run, not advice. "
"All arms of one comparison belong to one process "
"launch.")
containment_overridden: list = Field(default_factory=list)
warnings: list = Field(default_factory=list)
note: str = ""
class ExportOut(BaseModel):
bundle: str
origin_space_id: str
included: list
excluded: list = Field(
description="What was left out and why. Private packs land here "
"unless a human exports them from the shell.")
tensor_count: int = 0
bytes: int = 0
class ImportOut(BaseModel):
origin_space_id: str
origin_label: str = ""
landed: list
refused: list
index_dir: str
tensor_dir: str
# --------------------------------------------------------------- tools
@mcp.tool(description="Status of the local concept space: what evidence "
"exists, which other spaces have been imported, and "
"how much tensor data is on disk.")
def space_status() -> SpaceOut:
return SpaceOut(**api.space_status())
@mcp.tool(description="Check the space for inconsistencies: capture ids "
"that no longer recompute, tensor references that do "
"not resolve. Run this before trusting any number.")
def space_doctor() -> dict:
return api.space_doctor()
@mcp.tool(description="List records of one kind, local and imported, each "
"tagged with the space that measured it.")
def space_list(kind: str) -> dict:
return api.space_list(kind)
@mcp.tool(description="Start a new labelled corpus (an experiment design "
"table).")
def corpus_new(name: str, notes: str = "") -> dict:
return api.corpus_new(name, notes=notes)
@mcp.tool(description="Add one labelled observation to a corpus. Labels are "
"free-form; split is extract|validate|adversarial|"
"null; matched_group ties rows that differ in exactly "
"one factor; source_id is the shoot/take lineage, and "
"two rows sharing it are one observation however "
"different their bytes are.")
def corpus_add(name: str, source_kind: str, source_ref: str,
labels: Optional[dict] = None, split: str = "extract",
matched_group: Optional[str] = None,
source_id: Optional[str] = None,
notes: str = "") -> dict:
return api.corpus_add(name, source_kind, source_ref, labels=labels,
split=split, matched_group=matched_group,
source_id=source_id, notes=notes)
@mcp.tool(description="Corpus composition plus the protocol gaps it "
"carries. Read the warnings before planning a "
"capture, not after.")
def corpus_status(name: str) -> CorpusOut:
return CorpusOut(**api.corpus_status(name))
@mcp.tool(description="Which other labels move exactly with a target "
"label. The cheapest honest check available: if every "
"example of the target also carries some other "
"attribute, no factorization can separate them.")
def corpus_confounds(name: str, target: str) -> ConfoundOut:
return ConfoundOut(**api.corpus_confounds(name, target))
@mcp.tool(description="Cost and sanity-check a three-arm capture without "
"running anything. Returns forward count, tap points, "
"and every protocol warning. Launching the capture "
"itself is a human's call.")
def capture_plan(corpus: str, backend: str = "h3", store: str = "stats",
replicates: int = 1) -> PlanOut:
return PlanOut(**api.capture_plan(corpus, backend=backend, store=store,
replicates=replicates))
@mcp.tool(description="Assert that a control row really is shape-matched to "
"a reference row before spending the forwards. "
"Refuses on a mismatch naming the key; keys neither "
"row recorded come back as `unverifiable`, which is "
"silence and not agreement.")
def arms_check(corpus: str, r_row_id: str, n_row_id: str,
backend: str = "h3") -> dict:
return api.arms_check(corpus, r_row_id, n_row_id, backend=backend)
@mcp.tool(description="File a capture manifest the tap has already "
"produced. Runs no model: it recomputes the id from "
"the manifest's own fields and refuses a manifest "
"whose claimed id disagrees with them.")
def capture_record(manifest: dict) -> dict:
return api.capture_record(manifest)
@mcp.tool(description="Export the shareable part of this space to a "
"portable bundle. Packs marked private are held back "
"and reported in `excluded`.")
def space_export(out_path: str, note: str = "") -> ExportOut:
res = api.space_export(out_path, include_private=False, note=note)
return ExportOut(**{k: res[k] for k in
("bundle", "origin_space_id", "included", "excluded",
"tensor_count", "bytes")})
@mcp.tool(description="Import another space's bundle. Its records land in "
"their own namespace with their origin stamped and "
"never merge into your own measurements.")
def space_import(bundle_path: str) -> ImportOut:
return ImportOut(**api.space_import(bundle_path))
@mcp.tool(description="List concept packs, local and imported, with their "
"visibility and evidence status.")
def pack_list() -> dict:
return api.pack_list()
if __name__ == "__main__":
sys.exit(mcp.run())
|