PyCharter for AI agents¶

PyCharter is a toolkit for AI agents: the tools, grounding, and guardrails an external agent calls when it must be correct about your data. PyCharter never defines or runs an agent — your framework owns the model, the prompt, and the loop. PyCharter supplies three things:

Eyes — a just-in-time, governance-stripped schema for a contract, and governed concept context.
Conscience — graduated validation returned as model-consumable feedback, plus a bounded repair loop.
Hands (guarded) — a focused set of actions (propose and commit a contract version, propose a field↔concept binding) that are off by default, policy-scoped, approval-gated, and audit-logged.

Everything is available over MCP with zero pycharter install on the agent side, or in-process via the Python API and framework adapters.

The runtime-agent path (produce conformant data)¶

An agent emits a record; you validate it and hand back feedback until it conforms.

For a runnable, no-model example that pins an exact contract version and shows invalid feedback followed by an accepted repair, run this from the repository root:

python -m pip install -e '.[ai]'
python examples/agent_readiness/run_demo.py --transport local --domain support
python examples/agent_readiness/run_demo.py --transport mcp --domain support
python examples/agent_readiness/run_demo.py --transport local --domain energy

The same runner exercises a real stdio MCP session and an unrelated energy domain with the same prepare/verify API and read-only policy file. The full walkthrough is in examples/agent_readiness/README.md. These examples are contract-only and report recorded: false; they do not claim hosted preparation or durable receipts.

from pycharter import Validator
from pycharter.agent import repair_loop

validator = Validator.from_file("contracts/order.yaml")

# `emit` is YOUR agent: given feedback (None first) it returns a candidate.
def emit(feedback: str | None) -> dict:
    return call_your_model(schema=validator.to_llm_schema(), feedback=feedback)

result = repair_loop(emit, validator, max_attempts=3, on_exhaust="quarantine")
if result.is_valid:
    persist(result.data)          # coerced, contract-clean record
else:
    send_to_dlq(result.dead_letter)   # bounded, auditable failure

Validator.to_llm_schema(budget_tokens=...) — a pruned JSON Schema (enums and descriptions kept, governance/vendor keys stripped).
ValidationResult.to_agent_feedback() — a compact correction message; the offending value of an undeclared field is never echoed. When an undeclared field is a near-miss of a declared one, the line carries a hint (... did you mean 'email_address'?). The hint is advice only: the record is still rejected until the agent uses the declared name.
repair_loop / arepair_loop — validate → feedback → retry, escalating on exhaustion via raise, return, or quarantine.

The coding-agent path (author and evolve contracts)¶

Use the MCP tools (below) or the Python services directly to sketch a contract from a sample, classify a change as breaking or compatible, and lint it — all draft-only. When the change is ready, the agent proposes it and a human approves before it is committed (see Guarded writes).

Agent-readiness publication lint¶

Run the optional readiness rules before publishing a contract to an agent:

pycharter contract lint contracts/order.yaml --agent-readiness --fail-on warning

Set the --fail-on threshold in the team's CI configuration (info, warning, error, or never). The default lint command keeps its existing governance checks. Readiness lint reports missing or placeholder field descriptions, protected schema literals, unsupported validation rules, and unreviewed imported ODCS guidance. ODCS context constraints are labelled advisory; they do not become executable rules through import.

Approved binding state and current concept-scheme versions live outside the contract. Python callers can pass those facts to lint_agent_readiness(contract, approved_bindings=..., current_scheme_versions=...); the pure linter then reports unreviewed bindings and stale schemes. The offline CLI reports that approval is unavailable rather than treating an authored binding as approved.

Provider tool definitions¶

tool = validator.to_tool_schema("anthropic", name="submit_order",
                                description="Submit a customer order")
# -> {"name": ..., "description": ..., "input_schema": <pruned schema>}

Framework adapters¶

from pycharter.contrib.agents import contract_tool, contract_output_model

tool, check = contract_tool(order_contract, description="Submit an order")
verdict = check(payload)          # .is_valid / .data / .feedback

Order = contract_output_model(order_contract)   # a Pydantic AI output_type

MCP server (zero-install for the agent)¶

Point any MCP host at the server:

pip install "pycharter[ai]"
pycharter agent serve            # stdio MCP; --role / --workspace bind policy

Read/advise tools (policy-filtered):

Tool	Purpose
`list_contracts`	Discover contracts (namespace-scoped).
`fetch_contract_schema`	JIT LLM schema (or provider tool schema) for a contract.
`prepare_contract_context`	Exact active contract, enforced rules, and approved meanings when the server has a bound preparation scope.
`validate_record`	Validate a payload → `is_valid` + `agent_feedback`.
`sketch_contract`	Infer a candidate contract from samples (draft only).
`diff_contract_candidate`	Classify a candidate vs a stored version.
`lint_contract`	Governance lint over a candidate.
`fetch_concept_pack` / `explain_field`	Governed concept context for a field.
`list_pipelines` / `fetch_pipeline_config`	Stored pipeline configs (credentials masked).
`list_streams` / `fetch_stream_config`	Stored streaming worker configs (credentials masked).
`get_quality_score` / `list_drift_events`	A contract's quality score; recent drift events.
`get_dlq_summary`	Quarantine counts by reason/status/stage (never payloads).
`list_concept_schemes`	Vocabularies in the server's workspace.
`get_agent_conformance`	Quarantined agent output and quality score per contract.

Guarded-write tools (only with allow_write: true in the policy):

Tool	Purpose
`propose_contract_version`	Validate, lint, and diff a candidate; stage it and return a token. Writes nothing.
`commit_contract_version`	Commit an approved proposal as an immutable contract version.
`bind_field_to_concept`	File a pending field↔concept binding suggestion for a reviewer.

Inspect and dry-run from the shell:

pycharter agent tools                       # list the catalog + tiers
pycharter capabilities --kind agent_tool    # the registry entries (tiers are authoritative here)
pycharter agent tools --config agent-tools.yaml   # what this policy exposes
pycharter agent try contracts/order.yaml \
    --record '{"total_amount":"NaN"}' --schema   # offline, no model
pycharter agent prepare orders@1.0.0 --namespace team --local-only

For approved semantic meanings, start the MCP server with operator-bound --principal-id, --namespace, and --workspace (plus --role). The tool call cannot set identity. pycharter agent prepare accepts the same requester flags for a dry-run preview. See Prepared contract context for profiles, budgets, and the current hosted-mode limits.

Guarded writes¶

Writes are off until an operator turns them on with a policy file:

# agent-tools.yaml
agent:
  allow_write: true
  principal: claude-code
  role: reviewer

pycharter agent serve --config agent-tools.yaml

Every write is two-phase:

The agent calls propose_contract_version with a candidate. PyCharter validates, lints, and diffs it against the current latest version, and returns a token, the diff, and a new / compatible / breaking classification. The whole contract is compared — schema, rules, data quality, bindings and policies — and a change whose compatibility cannot be determined counts as breaking. Nothing is written yet.
A human approves: in session, when the MCP host gates the commit_contract_version call; or later, out of band, in the web UI (Contracts → Agents → Proposals) or with pycharter agent proposals approve <token> --as <name>.
commit_contract_version creates the new, immutable contract version.

Proposals expire after 24 hours and commit once. A proposal whose base version moved on is refused and must be re-proposed, so the diff a human approved is always the diff that lands. Set approval.require_distinct_approver: true to require that someone other than the proposer approves. Every step is recorded in the audit log. Full reference: Agent tool policy & guarded writes.

SDK callers that do not use MCP get the same tools and gates from pycharter.contrib.agents.catalog_tools.

Measuring conformance¶

To see how well agent output conforms over time, give the repair loop an observer and the contract's name:

from pycharter.agent import repair_loop
from pycharter.agent.observability import AgentConformanceObserver
from pycharter.dlq import DatabaseDeadLetterStore

observer = AgentConformanceObserver()   # needs prometheus_client ([server] extra)
observer.serve(9464)                    # http://127.0.0.1:9464/metrics
dlq = DatabaseDeadLetterStore(database_url)  # the pycharter database

result = repair_loop(
    emit, validator, on_exhaust="quarantine", observer=observer, contract="orders"
)
if result.dead_letter is not None:
    dlq.add_sync(result.dead_letter)

A Validator does not know its contract's name, so pass contract=. Without it, metrics and quarantined records are labelled unknown.

Live rates are Prometheus metrics:

Metric	Labels	Counts
`pycharter_agent_validations_total`	`contract`, `outcome` (`valid`, `invalid`)	Every validation of agent output.
`pycharter_agent_repairs_total`	`contract`, `outcome` (`conformed`, `exhausted`)	Finished repair loops.
`pycharter_agent_repair_attempts`	`contract`	Attempts per loop (histogram).

The MCP server counts its validate_record calls the same way when started with pycharter agent serve --metrics-port 9464.

Durable signals, which survive restarts, are summarized per contract by the get_agent_conformance tool, GET /api/v1/agent/conformance, and the web UI (Contracts → Agents → Conformance):

Quarantined: records a repair loop gave up on, by reason and by source. Only records stored in the database dead-letter queue are counted (the one the worker uses: PYCHARTER_DLQ_DATABASE_URL, falling back to the pycharter database). At most the 2,000 most recent records are scanned; truncated: true means the counts are a lower bound.
Quality: the contract's 0–100 quality score. Completeness comes from pipeline profiles and stability from drift events. It is absent when no pipeline profiled the contract in the window. When its source profile is restricted, unclassified, or small, quality_policy_gated: true distinguishes a withheld score from absent data; the score and component values are omitted.

Guardrails (why this is safe)¶

The stateless core never calls a model (CI-enforced). [ai] is optional; the default path works with a no-op provider.
Every MCP tool call is policy-filtered by role/workspace and never surfaces PII/restricted raw values.
Mutations (authoring writes) are a separate, fail-closed, approval-gated tier — off by default.