Align Artifacts¶
Align Artifacts — mvp.align_artifacts
Cluster: Safety & Alignment | Type: component | MCP Tools: 26
Overview¶
Versioned in-memory alignment artifact store for prompts, evaluation sets, specifications, and generic artifacts. Every store call appends a new version, preserving the full history; retrieve returns the latest version by default or a specific version by number. Artifacts are filterable by artifact_type and tags, and a search operation performs substring matching across name, content, and tags.
When to use:
- Maintaining a version-controlled library of alignment prompts and evaluation datasets during development
- Tracking how prompts or specs evolve across experiment iterations without a database
- Sharing a common artifact store across multiple pipeline components in a session
Example:
from mvp.align_artifacts import AlignArtifactsBlock, ArtifactsInput
block = AlignArtifactsBlock(name="artifacts")
block.infer(ArtifactsInput(op="store", name="safety_prompt", content="Do no harm.", artifact_type="prompt"))
result = block.infer(ArtifactsInput(op="retrieve", name="safety_prompt"))
# result.value.artifacts[0].version → 1; .content → "Do no harm."
Works well with: align_prompt_library, align_specs, align_evals
Production Readiness¶
align_artifacts is suitable for MVP and pilot workflows where a local MCP install needs a durable, inspectable reference library without external services. The SQLite-backed MCP block preserves content, metadata, and tag history across versions; tag-only changes are versioned; rollback restores prior tags; links require existing artifacts; and hard deletes clean up related versions and links.
Remaining caveats:
- SQLite is the launch default. It is appropriate for local installs and small pilot teams, but it is not a multi-tenant enterprise data layer by itself.
- Tags are stored as JSON text and searched with simple substring queries. This keeps install friction low, but large libraries should move to indexed tag tables or full-text search.
- Access control, tenant isolation, audit retention policy, encryption-at-rest configuration, and automated backup/restore should be handled by the deployment layer before using this as a shared customer system.
- Import and export are whole-store JSON operations intended for portability and backup, not high-volume synchronization.
artifacts_validateis a non-empty content check only; callers that require schema, format, or domain validation must apply those checks before trusting content.artifacts_capabilitiesis read-only and reports backend type, redacted DB path, WAL mode, counts, enabled ops,completion_state,warning_card, andevidencewithout reading secret values.- The in-memory
AlignArtifactsBlockis useful for tests and single-session pipelines. UseAlignArtifactsMCPBlockwith an explicitARTIFACTS_DB_PATHfor persistent user workflows.
Public API¶
AlignArtifactsBlock(AIBlock[ArtifactsInput, ArtifactsOutput, dict])¶
Versioned alignment artifact store.
| Field | Type | Default |
|---|---|---|
name | str | 'align_artifacts' |
resource_bounds | ResourceBounds \| None | None |
usage | ResourceUsage | field(default_factory=ResourceUsage) |
db_path | str \| None | None |
Methods:
infer(data: ArtifactsInput) -> Result[ArtifactsOutput]¶
verify_artifact_integrity(artifact_name: str, required_tags: list[str] | None = None, min_content_length: int = 1, required_type: str | None = None) -> Result[dict[str, Any]]¶
Verify an artifact meets integrity constraints.
ArtifactEntry(BaseModel)¶
| Field | Type | Default |
|---|---|---|
name | str | required |
content | str | required |
artifact_type | ArtifactType | 'other' |
tags | list[str] | Field(default_factory=list) |
version | int | 1 |
metadata | dict[str, str] | Field(default_factory=dict) |
source | str | '' |
downloaded_at | str | '' |
review_status | ArtifactProvenance | 'demo_seed' |
reviewer | str | '' |
confidence | float | 0.0 |
ArtifactsInput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | ArtifactOp | required |
name | str | '' |
content | str | '' |
artifact_type | ArtifactType | 'other' |
tags | list[str] | Field(default_factory=list) |
metadata | dict[str, str] | Field(default_factory=dict) |
query | str | '' |
version | int \| None | None |
source | str | '' |
downloaded_at | str | '' |
review_status | ArtifactProvenance | 'demo_seed' |
reviewer | str | '' |
confidence | float | 0.0 |
ArtifactsOutput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | str | required |
artifacts | list[ArtifactEntry] | Field(default_factory=list) |
count | int | 0 |
message | str | '' |
degraded | bool | False |
degradation_reason | str | '' |
completion_state | str | 'qualified-draft' |
warning_card | dict[str, str] | Field(default_factory=dict) |
evidence | dict[str, str] | Field(default_factory=dict) |
request_id | str | '' |
run_id | str | '' |
AlignArtifactsMCPBlock(AIBlock[MCPArtifactsInput, MCPArtifactsOutput, dict])¶
| Field | Type | Default |
|---|---|---|
name | str | 'align_artifacts_mcp' |
state | dict \| None | None |
db_path | str | ':memory:' |
resource_bounds | ResourceBounds \| None | None |
usage | ResourceUsage | field(default_factory=ResourceUsage) |
Methods:
infer(inp: MCPArtifactsInput) -> Result[MCPArtifactsOutput]¶
MCPArtifactsInput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | ArtifactMCPOp | required |
artifact_id | str \| None | None |
artifact_type | str \| None | None |
name | str \| None | None |
content | str \| None | None |
tags | list[str] | Field(default_factory=list) |
metadata | dict[str, Any] | Field(default_factory=dict) |
version | int \| None | None |
target_id | str \| None | None |
link_type | str \| None | None |
items | list[dict[str, Any]] | Field(default_factory=list) |
collection_id | str \| None | None |
collection_name | str \| None | None |
query | str \| None | None |
limit | int | 20 |
MCPArtifactsOutput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | str | required |
success | bool | required |
artifact_id | str \| None | None |
artifact | dict[str, Any] \| None | None |
artifacts | list[dict[str, Any]] | Field(default_factory=list) |
version | int \| None | None |
versions | list[dict[str, Any]] | Field(default_factory=list) |
links | list[dict[str, Any]] | Field(default_factory=list) |
collection_id | str \| None | None |
collection | dict[str, Any] \| None | None |
collections | list[dict[str, Any]] | Field(default_factory=list) |
count | int | 0 |
stats | dict[str, Any] | Field(default_factory=dict) |
message | str | '' |
error | str | '' |
degraded | bool | False |
degradation_reason | str | '' |
completion_state | str | 'qualified-draft' |
warning_card | dict[str, Any] | Field(default_factory=dict) |
evidence | dict[str, Any] | Field(default_factory=dict) |
request_id | str | '' |
run_id | str | '' |
MCP Tools¶
| Operation | Source |
|---|---|
store | artifacts_mcp |
retrieve | artifacts_mcp |
list | artifacts_mcp |
delete | artifacts_mcp |
search | artifacts_mcp |
update | artifacts_mcp |
get_version | artifacts_mcp |
list_versions | artifacts_mcp |
rollback | artifacts_mcp |
compare_versions | artifacts_mcp |
add_tag | artifacts_mcp |
remove_tag | artifacts_mcp |
list_by_tag | artifacts_mcp |
link | artifacts_mcp |
get_links | artifacts_mcp |
bulk_store | artifacts_mcp |
export_json | artifacts_mcp |
import_json | artifacts_mcp |
validate | artifacts_mcp |
create_collection | artifacts_mcp |
get_collection | artifacts_mcp |
list_collections | artifacts_mcp |
stats | artifacts_mcp |
info | artifacts_mcp |
list_patterns | artifacts_mcp |
capabilities | artifacts_mcp |