Plugin API Reference¶
This page documents executable generic plugin boundaries.
Agent Runtimes¶
adapter ¶
Claude Agent SDK runtime adapter.
This module owns the concrete Claude SDK import and event/message
normalization for the legacy research loop. BaseAgent remains as a
backward-compatible caller API, but the actual runtime execution lives here.
LegacyClaudeRuntimeOptions
dataclass
¶
Compatibility options used while legacy BaseAgent calls the Claude SDK runtime adapter.
LegacyAgentResult
dataclass
¶
Legacy-shaped terminal result returned to older research_loop callers.
ClaudeSdkAgentRuntime ¶
AgentRuntime adapter that drives Claude Code SDK and normalizes events for Praxist.
execute
async
¶
execute(
request: AgentRunRequest,
context: AgentRuntimeExecutionContext,
) -> AgentRunResult
Execute a normalized AgentRunRequest via the Claude SDK adapter.
execute_sync ¶
execute_sync(request: AgentRunRequest) -> AgentRunResult
Return a deterministic transcript for offline fixture runtime conformance.
claude_setting_sources_from_env ¶
claude_setting_sources_from_env() -> list[str]
Return the Claude Code settings scope for Praxist autonomous runs.
Praxist passes MCP servers, permissions, cwd, model, and credentials explicitly through the SDK options. Defaulting to project settings lets unrelated Claude Code project context compete with the Praxist research task and can cause a peer to wait for human instructions. Local settings preserve operator-local overrides without loading project/user memory by default.
is_billing_error ¶
is_billing_error(error_msg: str) -> bool
Classify whether an error string represents provider billing or quota failure.
format_legacy_message ¶
format_legacy_message(
message: Any, agent_name: str
) -> str | None
Format a Claude SDK stream message for legacy logs without leaking raw provider objects.
extract_legacy_output ¶
extract_legacy_output(
messages: list[Any],
) -> dict[str, Any]
Extract text, thinking blocks, tool counts, and usage from Claude SDK stream messages.
create_runtime ¶
create_runtime() -> ClaudeSdkAgentRuntime
Manifest entrypoint that constructs the Claude SDK runtime plugin.
adapter ¶
Official Codex Python SDK runtime backed by a long-lived app-server.
CodexSdkRuntime ¶
Execute independent Codex threads through shared app-server clients.
discover_managed_credential ¶
discover_managed_credential(
model_provider_ref: str,
) -> CredentialRef | None
Return a redacted reference for a saved native ChatGPT login.
execute
async
¶
execute(
request: AgentRunRequest,
context: AgentRuntimeExecutionContext,
) -> AgentRunResult
Run one request and stream typed SDK notifications into Praxist.
execute_sync ¶
execute_sync(request: AgentRunRequest) -> AgentRunResult
Return a deterministic transcript for offline conformance tests.
create_runtime ¶
create_runtime() -> CodexSdkRuntime
Return the process-local Codex SDK runtime plugin instance.
available_chatgpt_models ¶
available_chatgpt_models() -> tuple[str, ...]
Return model identifiers advertised for the saved ChatGPT account.
verify_chatgpt_model_available ¶
verify_chatgpt_model_available(model: str) -> str
Return the canonical model id or fail before a multi-peer launch.
API Providers¶
adapter ¶
Executable OpenRouter model provider plugin.
create_provider ¶
create_provider() -> ModelProviderAdapter
Manifest entrypoint for the OpenRouter model-provider adapter.
adapter ¶
Executable Anthropic Messages model provider plugin.
create_provider ¶
create_provider() -> ModelProviderAdapter
Manifest entrypoint for the Anthropic Messages model-provider adapter.
adapter ¶
Executable OpenAI-compatible model provider plugin.
create_provider ¶
create_provider() -> ModelProviderAdapter
Manifest entrypoint for generic OpenAI-compatible model providers.
adapter ¶
Executable DeepSeek OpenAI-compatible model provider plugin.
create_provider ¶
create_provider() -> ModelProviderAdapter
Manifest entrypoint for the DeepSeek OpenAI-compatible provider alias.
Workflow Stage¶
startup ¶
Startup/finalization bridge for the plugin-local research_loop backend.
ResearchLoopPluginRun
dataclass
¶
Prepared research-loop run bundle.
is_research_loop_task_project ¶
is_research_loop_task_project(
task_path: str | Path, workspace: Path | None = None
) -> bool
Return whether a task project can run with the research_loop workflow stage.
is_research_loop_plugin_task ¶
is_research_loop_plugin_task(
task_ref: str, workspace: Path | None = None
) -> bool
Compatibility shim for external task projects.
default_runtime_for_task ¶
default_runtime_for_task(
task_ref: str, runtime_ref: str | None = None
) -> str
Return the default agent runtime reference for a task descriptor.
default_model_provider_for_task ¶
default_model_provider_for_task(
task_ref: str, model_provider_ref: str | None = None
) -> str
Return the default model provider reference for a task descriptor.
default_budget_policy_for_task ¶
default_budget_policy_for_task(
task_ref: str, budget_policy_ref: str | None = None
) -> str
Return the default budget policy reference for a task descriptor.
prepare_research_loop_plugin_run ¶
prepare_research_loop_plugin_run(
*,
task_ref: str | None = None,
task_project_path: str | Path | None = None,
task_project: TaskProject | None = None,
workspace: Path,
run_dir: Path,
runtime_ref: str,
model_provider_ref: str,
budget_policy_ref: str,
model: str,
local_mode: bool,
frontier_strategy: str,
credential_profile: str | None = None,
command: str = "",
deprecated_args_seen: list[str] | None = None,
resolve_only: bool = False,
resume: bool = False,
resume_policy: str = "completed_generation",
) -> ResearchLoopPluginRun
Resolve task, plugins, credentials, budget, and startup artifacts.
finalize_research_loop_plugin_run ¶
finalize_research_loop_plugin_run(
prepared: ResearchLoopPluginRun,
*,
success: bool,
result: dict[str, Any] | None = None,
error: str | None = None,
exit_code: int | None = None,
) -> None
Write terminal run summary, materialized views, and trajectory events.
stage ¶
Executable research_loop workflow stage plugin.
ResearchLoopStageContext
dataclass
¶
Execution context passed from core workflow dispatch into the research_loop stage.
ResearchLoopStage ¶
WorkflowStage wrapper around the plugin-local GenerationLoop backend.
create_stage ¶
create_stage() -> ResearchLoopStage
Manifest entrypoint that constructs the research_loop workflow stage.
run_research_loop_stage ¶
run_research_loop_stage(
context: ResearchLoopStageContext,
) -> WorkflowStageResult
Compatibility entrypoint for running research_loop without a prebuilt stage object.
planned_research_loop_usage ¶
planned_research_loop_usage(
task_spec: Any,
) -> dict[str, float]
Return the canonical planned usage for grant and stage validation.
c5_materializer ¶
C5 materializers for legacy research_loop run directories.
The Step 14 boundary is deliberately conservative: legacy SQLite/YAML/graph files remain operational inputs, while this module imports their contents into canonical append-only ledgers and artifacts. The importer never deletes or rewrites legacy files.
LegacyResearchMemoryEntry
dataclass
¶
Canonicalized research-memory row imported from legacy YAML ledgers.
LegacyRunDirAdapter ¶
Read-only adapter over the legacy run_dir layout.
materialize_legacy_c5_views ¶
materialize_legacy_c5_views(
prepared: Any,
result: dict[str, Any],
*,
trajectory: Any,
artifacts: ArtifactWriter,
) -> dict[str, int]
Import legacy research_loop state into C5-compatible ledgers and artifact indexes.
adapter ¶
Local reviewer implementation for the optional reviewer workflow stage.
The reviewer checks claims against the run record that already exists on disk. It never re-runs task evaluators and never promotes or demotes candidates; the output is an audit artifact for operators and agents.
run_local_artifact_review ¶
run_local_artifact_review(
*,
run_dir: Path,
run_id: str,
stage_ref: str,
source_event_ids: list[str] | None = None,
) -> dict[str, Any]
Review artifact/provenance consistency and persist an audit report.
Tools¶
adapter ¶
Generalized evaluation MCP tools.
Replaces the W2S-specific evaluate_predictions / PGR tools with a task-agnostic metrics logging system.
Dual-mode
- Local mode (single server): reads/writes shared SQLite directly
- Server mode (multi-machine): POSTs to orchestrator HTTP API
adapter ¶
Finding Graph Query MCP tools — read-only navigation over the sidecar graph.
Per the finding-graph section of docs/concepts/architecture.md, v1 exposes three read-only tools:
get_finding_neighbors one-hop around a finding
get_finding_subgraph depth-limited subgraph, default depth=1
get_unlinked_recent_findings recent findings not yet in the graph
(protects exploration diversity)
These are ADVISORY. Edges are navigation, not conclusions. Raw findings in shared_findings/ remain the source of truth. If the graph is absent (maintainer disabled, DB locked, etc.) these tools return empty results gracefully — they never block the main MCP flow.
adapter ¶
adapter ¶
MCP tools for PI panel — read-only queries over research memory.
These tools are exposed ONLY to the PI synthesizer agents (Builder / Skeptic / Portfolio / External Validity / Chair). Peers (research workers) do NOT receive these tools.
The MCP server is created via create_memory_tools_server(run_dir)
which closes over the run_dir — the SDK tool handlers don't need to
take it as a parameter. This mirrors the registration pattern used
by frontier_tools / finding_graph_query.
get_evidence_card ¶
get_evidence_card(
run_dir,
evidence_id: str,
max_generation_id: int | None = None,
) -> dict[str, Any]
Return full evidence card by id, or {error: ...} if not found.
Use this when the role-specific pack you received omitted details for a card you need to inspect more closely.
query_evidence_cards ¶
query_evidence_cards(
run_dir,
claim_id: str | None = None,
mechanism: str | None = None,
peer_id: str | None = None,
generation_id: int | None = None,
is_negative: bool | None = None,
limit: int = 20,
max_generation_id: int | None = None,
) -> list[dict[str, Any]]
Filter evidence cards. Returns id + short interpretation only.
To get a full card, follow up with get_evidence_card(evidence_id).
query_coverage_matrix ¶
query_coverage_matrix(
run_dir,
variant_family: str | None = None,
parameter: str | None = None,
bridge_pair: list[str] | None = None,
bridge_dimension: str | None = None,
max_generation_id: int | None = None,
) -> dict[str, Any]
Check if (variant_family, parameter) grid or (pair, dimension) bridge is covered.
MUST be called by Bridge contracts before assignment. Returns {covered: bool, points: [...], sources: [...], or empty}.
list_active_claims ¶
list_active_claims(
run_dir, max_generation_id: int | None = None
) -> list[dict[str, Any]]
List active claims with status, confidence, boundary, and supports/challenges counts.
list_open_objections ¶
list_open_objections(
run_dir, max_generation_id: int | None = None
) -> list[dict[str, Any]]
List open / experiment-assigned dissent entries.
get_ledger_entry ¶
get_ledger_entry(
run_dir,
ledger_name: str,
entry_id: str,
max_generation_id: int | None = None,
) -> dict[str, Any]
Read a single entry from any ledger.
Supported ledger_names: claim_ledger, hypothesis_ledger, mechanism_ledger, coverage_matrix, negative_evidence_ledger, retired_claim_ledger, dissent_ledger, frontier_delta_ledger, role_roi_ledger.
resolve_source_ref ¶
resolve_source_ref(
run_dir,
source_ref: dict[str, Any],
max_generation_id: int | None = None,
) -> dict[str, Any]
Lazy-load the raw file behind an evidence_card.source_ref.
Use SPARINGLY: this loads the original JSON / YAML, which is verbose. Prefer get_evidence_card for normal operation.
create_memory_tools_server ¶
create_memory_tools_server(run_dir)
Create an MCP server exposing the memory query tools to PI agents.
The PI panel attaches this server when multi_pi.enabled=True. Peers do NOT see this server (their allowed_tools list does not include mcp__memory-tools__* entries).
create_tool_plugin ¶
create_tool_plugin() -> dict[str, object]
Manifest entrypoint that exposes PI research-memory query tools.
adapter ¶
MCP tool for downloading prior work snapshots.
Agents can request specific workspace snapshots to reference or build on.
adapter ¶
No-key public literature/database/open-access lookup tools.
This adapter completes the long-standing tool_server:literature_lookup
contract without introducing new credentials. It intentionally stays small:
search a few public sources, normalize records, and degrade per source when a
remote endpoint is unavailable. External literature is contextual evidence for
research planning, not task evaluation truth.
literature_search ¶
literature_search(
query: str,
sources: str
| list[str]
| tuple[str, ...]
| None = None,
max_results: int = _DEFAULT_MAX_RESULTS,
*,
http_client_factory: Any = None,
clock: Any = None,
) -> dict[str, Any]
Search public literature sources and return normalized records.
scientific_database_search ¶
scientific_database_search(
query: str,
sources: str
| list[str]
| tuple[str, ...]
| None = None,
max_results: int = _DEFAULT_MAX_RESULTS,
*,
http_client_factory: Any = None,
) -> dict[str, Any]
Search no-key public scientific databases beyond paper indexes.
This is deliberately read-only and compact. It is meant to help task initialization and literature-scout roles find authoritative entity, trial, and biomedical records without adding API-key requirements.
literature_open_access_text ¶
literature_open_access_text(
identifier_or_url: str,
max_chars: int = _DEFAULT_MAX_TEXT_CHARS,
*,
http_client_factory: Any = None,
clock: Any = None,
) -> dict[str, Any]
Fetch open-access text or PDF bytes metadata with provenance.
The function never tries to bypass paywalls. DOI/arXiv/OpenAlex/PubMed
identifiers are first resolved to public metadata, then the best open URL
is fetched when one is available. PDF responses are recorded with content
hash and size; OCR/text extraction remains delegated to pdf_reader.
literature_resolve ¶
literature_resolve(
identifier: str,
*,
http_client_factory: Any = None,
clock: Any = None,
) -> dict[str, Any]
Resolve a DOI, PMID, arXiv ID, or OpenAlex ID into one normalized record.
literature_source_guide ¶
literature_source_guide(
domain: str = "", objective: str = ""
) -> dict[str, Any]
Return source-selection guidance for task-local research context gathering.
handle_literature_search ¶
handle_literature_search(
args: dict[str, Any],
) -> dict[str, Any]
Manifest handler wrapper for direct tool execution.
handle_literature_resolve ¶
handle_literature_resolve(
args: dict[str, Any],
) -> dict[str, Any]
Manifest handler wrapper for direct tool execution.
handle_literature_open_access_text ¶
handle_literature_open_access_text(
args: dict[str, Any],
) -> dict[str, Any]
Manifest handler wrapper for direct tool execution.
handle_scientific_database_search ¶
handle_scientific_database_search(
args: dict[str, Any],
) -> dict[str, Any]
Manifest handler wrapper for direct tool execution.
handle_literature_source_guide ¶
handle_literature_source_guide(
args: dict[str, Any],
) -> dict[str, Any]
Manifest handler wrapper for direct tool execution.
create_literature_lookup_server ¶
create_literature_lookup_server() -> Any
Create the MCP server exposing public literature lookup tools.
create_tool_plugin ¶
create_tool_plugin() -> dict[str, object]
Manifest entrypoint exposing the literature lookup tool server descriptor.
Graph Maintainer¶
adapter ¶
Executable finding graph maintainer plugin.
engine ¶
Finding Graph rule engine + health metrics.
Implements the sidecar graph index over shared findings specified in the
finding-graph section of docs/concepts/architecture.md.
Philosophy (quoting the doc): - edges are navigation, not conclusions - conservative first: prefer related_to over strong edges when unsure - do not merge nodes — originals remain source of truth - rules only; no LLM calls in v1
Runtime: deterministic Python. Safe to run in shadow mode alongside the orchestrator without affecting findings writes or frontier behavior.
FindingGraphBuilder ¶
Stateless rule engine.
Usage
builder = FindingGraphBuilder(all_findings) for f in builder.chronological(): proposed = builder.propose_edges_for(f) # insert_edges_batch(proposed)
One-shot backfill
edges = builder.build_all_edges()
chronological ¶
chronological() -> list[dict[str, Any]]
Yield findings in timestamp-ascending order (stable on ties by id).
propose_edges_for ¶
propose_edges_for(
new_finding: dict[str, Any],
) -> list[dict[str, Any]]
Run all rules against new_finding, return resolved edge list.
The returned edges always have src_finding_id = new_finding["id"]
(i.e. new → old; see design §3.1).
build_all_edges ¶
build_all_edges() -> list[dict[str, Any]]
Iterate all findings in chronological order and propose edges.
The edges are NOT deduped against SQLite here — caller should use
insert_edges_batch which silently skips duplicates via UNIQUE.
FindingGraphMaintainer ¶
Background daemon that periodically runs the rule engine over new findings and upserts their edges into SQLite.
Mirrors the FindingsSync shape (start/stop/sync_once) for a consistent orchestrator lifecycle.
sync_once ¶
sync_once() -> dict[str, Any]
Run one maintainer cycle. Returns a small summary dict.
Safe to call from multiple threads — concurrent calls no-op
(return status=busy) rather than duplicating work. Use
sync_once_blocking() when the caller NEEDS the cycle to
run (e.g. inter-generation barrier).
sync_once_blocking ¶
sync_once_blocking(
timeout: float = 300.0,
) -> dict[str, Any]
Run a cycle, waiting for any in-progress cycle to finish first. Used at generation boundaries where the orchestrator needs the graph to absorb the just-finished generation's findings before the next generation's prompts render — silently skipping would leave gen N+1 peers reading stale edges.
Returns {"status": "timeout"} if the lock cannot be
acquired within timeout seconds; this caps the worst-case
delay at a generation boundary.
reset_graph_observability_state ¶
reset_graph_observability_state() -> None
Zero the module-level counters and status. Intended for FindingGraphMaintainer.init (fresh-run contract) and for test harnesses that spin up multiple orchestrators in one Python process. Without this, counters monotonically accumulate across logically-independent runs.
compute_graph_health ¶
compute_graph_health() -> dict[str, Any]
Snapshot of graph size + coverage + edge-type distribution.
Intended to be written each maintainer cycle to run_dir/graph/graph_health.json.
write_graph_health ¶
write_graph_health(out_dir: Path) -> dict[str, Any]
Compute health + atomic-write to <out_dir>/graph_health.json.
build_session_start_graph_context ¶
build_session_start_graph_context(
peer_id: str,
max_prior_findings: int = 5,
max_neighbors: int = 6,
max_anchors: int = 4,
) -> str
Markdown snippet summarizing the graph neighborhood most relevant to this peer's starting session.
Returns a string that either (a) lists the top N cross-peer neighbors of this peer's recent findings (for a peer that's been active), or (b) lists the current graph's orientation anchors (for a fresh peer). Returns empty string on any failure — the caller treats this as optional.
cli ¶
CLI: build the Finding Graph index for a run.
Modes
--mode backfill Run rule engine over all findings in the run's SQLite,
upsert edges. Idempotent via UNIQUE(src, dst, type).
--mode health Compute + print graph health stats; write graph_health.json.
--mode daemon Start FindingGraphMaintainer and block (shadow mode).
Orchestrator already does this automatically in local mode;
this is for ad-hoc / post-hoc usage.
--mode viz Render
Examples:
# Backfill edges for a completed run
python -m praxist.plugins.graph_maintainers.finding_graph_mvp.cli \
--run-dir
# Just check health of an already-built graph
python -m praxist.plugins.graph_maintainers.finding_graph_mvp.cli \
--run-dir
Design reference: the finding-graph section of docs/concepts/architecture.md.
cmd_backfill ¶
cmd_backfill(args)
Backfill finding graph edges from the current SQLite findings store.