coact

coact — reuse your AI stuff across the layers of the modern agent stack.

coact (“co-act”: skills and agents acting as one reusable substrate) owns the two transitions the rest of the ecosystem doesn’t:

python functions/scripts  →  .claude/skills/  →  .claude/agents/  →  running agents
        (py2mcp, aw)            (skill pkg)         COMPLETE (coact)    REALIZE (coact)

• COMPLETE lifts .claude/skills/ into .claude/agents/ definitions, adding the agent-only “extras” (persona, return contract, tool allowlist, model, memory).

• REALIZE turns a completed definition into an actually running agent, picking an execution backend (host agent / Agent SDK / MCP-exposed).

It is glue over skill (foundation), aw (runtime substrate), and py2mcp (tool exposure) — see misc/docs/REUSE.md.

Simple usage:

from coact import complete, emit_agent, realize

agent = complete('.claude/skills/ux-analyst')      # Skill -> AgentDefinition
emit_agent(agent, 'claude-agents-md', dest='.claude/agents/')
realize(agent, backend='host')                     # materialize for Claude Code

class coact.AgentDefinition(name: str, description: str, prompt: str = '', tools: list[str] | None = None, disallowed_tools: list[str] = <factory>, model: Literal['sonnet', 'opus', 'haiku', 'inherit'] | None=None, skills: list[str] = <factory>, memory: Literal['user', 'project', 'local'] | None=None, mcp_servers: list[Any] = <factory>, permission_mode: str | None = None, returns: ReturnContract = <factory>, consumes: str | None = None, source_skill: str | None = None)

The SSOT: one object, two serializations (filesystem md + SDK form).

Fields mirror the host subagent schema (snake_case here; coact owns the mapping to the host’s camelCase frontmatter names in coact.emit), plus the coact-specific returns / consumes. tools=None means inherit all tools; an empty list means no tools — the distinction is preserved.

>>> ad = AgentDefinition(name='ux-analyst', description='Analyze UX bundles.')
>>> ad.name, ad.tools, ad.returns.is_empty()
('ux-analyst', None, True)

source_skill: str | None = None

Canonical key/name of the source skill this agent was completed from.

class coact.AgentPlan(agent: AgentDefinition, provenance: list[FieldProvenance] = <factory>, warnings: list[str] = <factory>)

The inspectable result of plan_completion: the agent + its provenance.

Shows the proposed AgentDefinition and, for each field, why it has its value — so a user can review before any file is written or agent spawned (progressive disclosure: dry-run first).

render() → str

Render a human-readable provenance table for terminal display.

>>> plan = AgentPlan(
...     agent=AgentDefinition(name='x', description='y'),
...     provenance=[FieldProvenance('model', 'sonnet', 'policy', 'worker')],
... )
>>> 'model' in plan.render() and 'policy' in plan.render()
True

class coact.AgentStore(root: Path | str | None = None, *, scope: str = 'project', project_dir: Path | str | None = None)

A MutableMapping[str, AgentDefinition] over a .claude/agents/ dir.

Keys are agent names; each agent is one <name>.md file.

>>> import tempfile
>>> from coact.base import AgentDefinition
>>> store = AgentStore(root=tempfile.mkdtemp())
>>> store['ux'] = AgentDefinition(name='ux', description='Analyze.', prompt='You are...')
>>> list(store)
['ux']
>>> store['ux'].description
'Analyze.'
>>> del store['ux']
>>> len(store)
0

class coact.CoactMeta(tools: list[str] | None = None, disallowed_tools: list[str] = <factory>, model: str | None = None, memory: str | None = None, permission_mode: str | None = None, skills: list[str] = <factory>, mcp: list[dict] = <factory>, returns: dict | None = None, consumes: str | None = None, persona: str | None = None)

The parsed coact: block (every field optional; all default to absent).

>>> m = CoactMeta.from_frontmatter({'coact': {'model': 'opus', 'tools': ['Read']}})
>>> m.model, m.tools, m.is_empty()
('opus', ['Read'], False)
>>> CoactMeta.from_frontmatter({'name': 'x'}).is_empty()
True

classmethod from_frontmatter(meta: dict) → CoactMeta

Build from a full frontmatter dict (reads the coact sub-mapping).

is_empty() → bool

True when the block carries no author-provided information.

return_contract() → ReturnContract

The author-pinned ReturnContract, if any.

class coact.CompletionPolicy(default_model: Literal['sonnet', 'opus', 'haiku', 'inherit'] = 'sonnet', default_memory: Literal['user', 'project', 'local'] | None = None, default_tools: tuple[str, ...] = ('Read', 'Grep', 'Glob'), read_only_tools: frozenset[str] = frozenset({'Glob', 'Grep', 'NotebookRead', 'Read', 'WebFetch', 'WebSearch'}), opus_keywords: tuple[str, ...] = ('orchestrat\\w*', 'architect\\w*', 'coordinat\\w*', '\\bplan(?:ning|s|ned)?\\b', '\\bdesign(?:s|ing|ed)?\\b', 'review\\w*'), opus_skill_threshold: int = 3, write_keywords: tuple[str, ...] = ('write', 'edit', 'create file', 'modify', 'generate', 'refactor'))

Injectable defaults for the §3.2 extras COMPLETE must stamp on a skill.

>>> p = CompletionPolicy()
>>> p.choose_model(tools=['Read', 'Grep'], description='Audit the bundle.', n_skills=1)
('haiku', 'read-only/explore: tools ⊆ read-only set')
>>> p.choose_model(tools=['Read', 'Write'], description='Implement the fix.', n_skills=1)[0]
'sonnet'
>>> p.choose_model(tools=['Read'], description='Orchestrate the review.', n_skills=1)[0]
'opus'

choose_memory() → tuple[Literal['user', 'project', 'local'] | None, str]

The default memory scope (opt-in; None = no memory).

choose_model(*, tools: list[str] | None, description: str, n_skills: int) → tuple[Literal['sonnet', 'opus', 'haiku', 'inherit'], str]

Route to a model from the effective tools / description / skill count.

infer_tools(skill: Skill, coact_meta: CoactMeta) → tuple[list[str], str]

Derive the tool allowlist (declared-or-heuristic + report; DECISIONS D3).

Precedence: an explicit coact: tools wins; otherwise a conservative heuristic from the skill’s resources/body. Never silently guesses — the reason is returned for provenance.

>>> from skill.base import Skill, SkillMeta
>>> s = Skill(meta=SkillMeta(name='x', description='y'), body='Write the report.')
>>> tools, why = CompletionPolicy().infer_tools(s, CoactMeta())
>>> 'Write' in tools and 'Read' in tools
True

opus_keywords: tuple[str, ...] = ('orchestrat\\w*', 'architect\\w*', 'coordinat\\w*', '\\bplan(?:ning|s|ned)?\\b', '\\bdesign(?:s|ing|ed)?\\b', 'review\\w*')

Regex patterns (matched case-insensitively against the description) that route to opus. Word-bounded so ‘plan.’ matches but ‘designer’/’designate’ don’t. Data, not code — extend or replace per policy.

override(**kwargs) → CompletionPolicy

Return a new policy with overridden fields (cascading defaults).

>>> CompletionPolicy().override(default_model='opus').default_model
'opus'

write_keywords: tuple[str, ...] = ('write', 'edit', 'create file', 'modify', 'generate', 'refactor')

body substrings that imply the agent writes/edits files.

class coact.FieldProvenance(field: str, value: Any, source: Literal['skill', 'coact-frontmatter', 'policy', 'inferred', 'synthesized-template', 'synthesized-llm', 'default'], note: str = '')

Where one AgentDefinition field’s value came from (dry-run audit).

>>> FieldProvenance('model', 'sonnet', 'policy', 'default worker').source
'policy'

class coact.IntegrationSpec(name: str, description: str = '', version: str = '0.1.0', tools: list[str] = <factory>, resources: list[str] = <factory>, prompts: list[str] = <factory>, tool_specs: list[ToolSpec] = <factory>, instructions: str | None = None, auth: str = 'none', deployment: str = 'local-stdio', author: str | None = None, source: str | None = None)

Target-neutral description of an integration to publish.

The connectivity core maps onto MCP’s three primitives. Tools come in two shapes that coexist: bare 'module:function' refs in tools (the mechanical/code ingress) and richer ToolSpec descriptors in tool_specs (the NL ingress, landscape-doc §9.1). resources/prompts and the auth/deployment hints are declared now (open-closed) for the remote connector and other targets to come.

>>> spec = IntegrationSpec(name='paths', tools=['os.path:basename'])
>>> spec.name, spec.tools, spec.deployment
('paths', ['os.path:basename'], 'local-stdio')
>>> IntegrationSpec(name='empty').is_empty()
True
>>> draft = IntegrationSpec(name='wx', tool_specs=[ToolSpec(name='get')])
>>> draft.is_empty(), draft.runnable_refs()
(False, [])

is_empty() → bool

True when there is nothing to publish (no tools/tool_specs/resources/prompts).

render() → str

A terminal-friendly summary of the (possibly draft) integration.

runnable_refs() → list[str]

The importable 'module:function' refs that back runnable tools.

Bare refs in tools plus the handler of every bound ToolSpec, de-duplicated preserving order. A draft whose tools are all proposed (unbound) returns [] — it has nothing a server can actually run yet.

class coact.PublishResult(target: str, dry_run: bool = False, artifact: Path | None = None, files: dict = <factory>, instructions: str = '', warnings: list[str] = <factory>)

The outcome of publish() — what was (or would be) produced.

dry_run mirrors realize(backend='host', dry_run=True): in a dry run artifact is None and files holds the would-write bundle members (relpath → short preview), so you can look before you leap.

>>> PublishResult(target='claude-local-mcpb', dry_run=True).render().splitlines()[0]
"Would publish (target='claude-local-mcpb', dry-run)"

render() → str

A terminal-friendly summary (used by the CLI).

class coact.RealizedHost(agents: dict[str, ~pathlib.Path]=<factory>, skills: dict[str, ~pathlib.Path]=<factory>, agents_dir: Path | None = None, skills_dir: Path | None = None, warnings: list[str] = <factory>, dry_run: bool = False)

The materialized result of host realization (files the host will discover).

When produced by a dry_run the paths are what would be written/linked (dry_run=True is recorded so the caller / CLI can label the preview); nothing on disk is touched.

class coact.ReturnContract(json_schema: dict = <factory>, ref: str | None = None, description: str = '')

The agent’s return contract: its final-message schema + a human summary.

Skills return nothing; agents must define a consumable output. The canonical form is a JSON Schema dict (portable across realization backends, DECISIONS D6). ref records a 'module:Name' source if the schema was resolved from a Pydantic model / dataclass / TypedDict.

>>> rc = ReturnContract(json_schema={'type': 'object'}, description='findings')
>>> rc.is_empty()
False
>>> ReturnContract().is_empty()
True

classmethod from_dict(d: dict | None) → ReturnContract

Parse from a coact: returns mapping (accepts schema_ref or json_schema).

>>> ReturnContract.from_dict({'schema_ref': 'ov.schemas:UxFindings'}).ref
'ov.schemas:UxFindings'

is_empty() → bool

True when no schema (inline or ref) has been set.

schema() → dict

Return the canonical JSON Schema, resolving ref if needed (else {}).

Honors DECISIONS D6 (“JSON Schema is canonical; refs resolve to it”): an inline json_schema wins; otherwise a module:Name ref is resolved best-effort to a schema. Returns {} when neither is available.

>>> ReturnContract(json_schema={'type': 'object'}).schema()
{'type': 'object'}
>>> 'properties' in ReturnContract(ref='coact.base:ReturnContract').schema()
True

to_dict() → dict

Serialize to a plain dict (for the agent-md coact: block), omitting empties.

class coact.RunnableAgent(agent_def: AgentDefinition, llm: Any = None, runner: Callable[[str, Any], Any] | None = None, default_tools: tuple[str, ...] = ('Read', 'Grep', 'Glob'), return_mode: str = 'auto')

An aw.AgenticStep-compatible runnable backed by the Claude Agent SDK.

Satisfies execute(input_data, context) -> (artifact, info_dict) so it drops into aw workflows and reuses aw’s retry/validation/human-in-loop. The SDK is imported lazily; runner is injectable so the agent can be constructed and unit-tested without a live API call (dependency injection).

build_options() → Any

Construct ClaudeAgentOptions from the agent definition (no API call).

execute(input_data: Any, context: Any = None) → tuple[Any, dict[str, Any]]

Run the agent over input_data; return (artifact, info) (aw protocol).

return_mode: str = 'auto'

'auto' (native output_format when the SDK supports it, else the forced return tool), 'output_format', or 'tool' (force the return_result tool — for older SDKs / extended thinking that cannot honor output_format). See DECISIONS D6.

Type:

How the return contract is realized

class coact.RunnableCrewAIAgent(agent_def: ~coact.base.AgentDefinition, model_map: dict = <factory>, default_model: str = 'openai/gpt-4o-mini', runner: ~collections.abc.Callable[[...], ~typing.Any] | None = None, tools_map: dict | None = None, use_response_format: bool = True)

An aw.AgenticStep-compatible runnable backed by a single crewai.Agent.

The Agent is built lazily and cached; runner is injectable so the agent runs in tests with no API key and no crewai install (dependency injection). The Agent is exposed via build_agent() / the agent property.

>>> from coact import AgentDefinition
>>> ad = AgentDefinition(name='x', description='d', prompt='You are X.', model='sonnet')
>>> RunnableCrewAIAgent(ad).resolve_model()
'anthropic/claude-sonnet-4-5'

property agent: Any

The native crewai.Agent — add it to your own Crew(agents=[...]).

build_agent() → Any

Build (once) and return the framework-native crewai.Agent.

Pure with respect to API calls — constructing the Agent does not invoke a model. Imports crewai lazily (checked first); only ever called on the default (non-injected) run path.

build_backstory() → str

The CrewAI backstory (persona + the D6 return-contract instruction).

build_goal() → str

The CrewAI goal (required, non-empty).

build_response_model() → type | None

Synthesize the pydantic class for kickoff(response_format=), or None.

None when there is no schema, use_response_format is off, or the schema is not flat enough to represent (then coact relies on the prompt instruction).

build_role() → str

The CrewAI role (required, non-empty) — the agent’s name.

execute(input_data: Any, context: Any = None) → tuple[Any, dict[str, Any]]

Run the agent over input_data; return (artifact, info) (aw protocol).

context is accepted for aw.AgenticStep compatibility and ignored. Uses the native .pydantic result when present; otherwise JSON-parses .raw (graceful, like litellm). An injected runner keeps crewai unimported.

resolve_model() → str

Map the definition’s model selector to a LiteLLM model string (slash form).

runner: Callable[[...], Any] | None = None

runner(*, agent, input_text, response_format) -> output (.raw / .pydantic); defaults to _default_crewai_runner(). The DI seam (mirrors litellm completion=).

tools_map: dict | None = None

{tool_name: callable | crewai BaseTool} — opt-in tool binding (D12); names with no entry are reported in info['warnings'] and never passed on.

use_response_format: bool = True

Also request native structured output (synthesized pydantic class) when possible.

class coact.RunnableLLMAgent(agent_def: ~coact.base.AgentDefinition, model_map: dict = <factory>, default_model: str = 'openai/gpt-4o-mini', completion: ~collections.abc.Callable[[...], ~typing.Any] | None = None, use_response_format: bool = True)

An aw.AgenticStep-compatible runnable backed by LiteLLM (any provider).

The return contract is realized two ways at once (belt-and-suspenders, since provider support for structured output varies): the JSON Schema is passed as LiteLLM’s response_format and embedded as an instruction in the system message. completion is injectable so the agent runs in tests with no API key; LiteLLM is imported lazily only on the default path.

>>> from coact import AgentDefinition, ReturnContract
>>> ad = AgentDefinition(name='x', description='d', prompt='You are X.', model='sonnet')
>>> RunnableLLMAgent(ad).resolve_model()
'anthropic/claude-sonnet-4-5'

build_kwargs(input_data: Any) → dict

Build the litellm.completion kwargs (model, messages, response_format).

build_messages(input_data: Any) → list[dict]

Build the chat messages: persona (+ return-contract instruction) then input.

completion: Callable[[...], Any] | None = None

completion(**kwargs) -> response; defaults to litellm.completion.

execute(input_data: Any, context: Any = None) → tuple[Any, dict[str, Any]]

Run the agent over input_data; return (artifact, info) (aw protocol).

context is accepted for aw.AgenticStep compatibility and ignored by this backend. If a provider rejects the structured response_format, the call is retried once without it — the schema is still requested via the system-prompt instruction (the belt-and-suspenders fallback made functional).

resolve_model() → str

Map the definition’s model selector to a LiteLLM model string.

A mapped selector wins; an explicit LiteLLM string in model is used verbatim; otherwise default_model. The lookup covers any hashable selector (so a model_map entry always applies if present).

use_response_format: bool = True

Also pass the schema as LiteLLM response_format (in addition to the prompt).

class coact.RunnableLLMGraphAgent(agent_def: ~coact.base.AgentDefinition, model_map: dict = <factory>, default_model: str = 'openai:gpt-4o-mini', factory: ~collections.abc.Callable[[...], ~typing.Any] | None = None, runner: ~collections.abc.Callable[[~typing.Any, ~typing.Any], ~typing.Any] | None = None, tools_map: dict | None = None, use_response_format: bool = True, structured_strategy: str = 'auto')

An aw.AgenticStep-compatible runnable backed by a LangGraph graph.

The compiled graph is built lazily and cached; factory is injectable so the agent can be constructed and unit-tested without a live API call or a langchain/langgraph install (dependency injection). The graph itself is exposed via build_agent() / the agent property for composition.

>>> from coact import AgentDefinition
>>> ad = AgentDefinition(name='x', description='d', prompt='You are X.', model='sonnet')
>>> RunnableLLMGraphAgent(ad).resolve_model()
'anthropic:claude-sonnet-4-5'

property agent: Any

The native CompiledStateGraph — add it as a node to your own StateGraph.

build_agent() → Any

Build (once) and return the framework-native CompiledStateGraph.

Pure with respect to API calls — constructing the graph does not invoke a model. The default path imports langchain/langgraph lazily and checks they are installed; an injected factory bypasses both.

build_response_format() → Any

Build the create_agent response_format for the return contract, or None.

Wraps the canonical JSON-Schema dict in ToolStrategy (default) or ProviderStrategy — both accept the raw dict. When langchain is not importable (an injected factory in tests, or the create_react_agent fallback, which accepts a bare dict), the raw schema dict is returned so the offline path needs no framework.

build_state(input_data: Any) → dict

Build the LangGraph input state (a single user message).

build_system_prompt() → str

Persona (+ the D6 return-contract instruction when a schema is declared).

execute(input_data: Any, context: Any = None) → tuple[Any, dict[str, Any]]

Run the agent over input_data; return (artifact, info) (aw protocol).

context is accepted for aw.AgenticStep compatibility and ignored. Uses the native structured_response when the graph returns one; otherwise falls back to JSON-parsing the final message text (graceful, like litellm) — langchain may omit structured_response even when a schema was requested.

factory: Callable[[...], Any] | None = None

factory(*, model, tools, system_prompt, response_format, name) -> graph; defaults to _default_langgraph_factory(). The primary DI seam.

resolve_model() → str

Map the definition’s model selector to a langchain provider:model string.

A mapped selector wins; an explicit provider string in model is used verbatim; otherwise default_model.

runner: Callable[[Any, Any], Any] | None = None

runner(graph, state) -> result.

Type:

Optional test-only bypass of graph.invoke

structured_strategy: str = 'auto'

'auto' / 'tool' -> ToolStrategy (safest cross-provider); 'provider' -> ProviderStrategy (native provider structured output).

tools_map: dict | None = None

{tool_name: callable | BaseTool} — opt-in tool binding (D12); names with no entry here are reported in info['unbound_tools'] and never passed on.

use_response_format: bool = True

Also request native structured output (in addition to the prompt instruction).

class coact.ToolSpec(name: str, description: str = '', input_schema: dict | None = None, handler: str | None = None)

A richer, target-neutral description of one tool in an IntegrationSpec.

The mechanical ingress represents tools as bare 'module:function' ref strings (IntegrationSpec.tools). The NL ingress (coact.nl_ingress) and the landscape-doc §9.1 model need more — a name, a description, an input JSON Schema, and an optional handler ref:

• A ToolSpec with a handler is bound (runnable): its ref joins the spec’s runnable set and the published server can import and call it.

• A ToolSpec without a handler is a proposed tool — a design draft to bind to real code (or supply a ref for) before it can run.

>>> ToolSpec(name='get_weather', handler='wx.api:current').is_bound()
True
>>> ToolSpec(name='get_weather').is_bound()
False

is_bound() → bool

True when a module:function handler backs this tool (so it can run).

coact.agents_dir(*, scope: str = 'project', project_dir: Path | str | None = None) → Path

Resolve the .claude/agents directory for the given scope.

scope='project' resolves against the detected project root (or project_dir); scope='global' resolves against ~/.claude.

>>> agents_dir(scope='global').as_posix().endswith('.claude/agents')
True

coact.back(agent: object) → Skill

Best-effort, lossy agent→skill extraction (harvest an agent into a skill).

Strips the persona/return-contract envelope down to a reusable skill stub (name + description + a pointer to the skills it referenced). The actual procedure usually lives in those referenced skills, so this cannot recover it — the result is a starting point the user fleshes out, not a faithful inverse of COMPLETE.

>>> from coact import AgentDefinition
>>> sk = back(AgentDefinition(name='ux', description='Analyze.', skills=['ux', 'shared']))
>>> sk.meta.name
'ux'
>>> 'lossy' in sk.body.lower()
True

coact.complete(source: str | Path | Skill, *, policy: CompletionPolicy | None = None, llm: object = None) → AgentDefinition

Complete a skill into an AgentDefinition (the plan’s agent).

>>> from skill.base import Skill, SkillMeta
>>> s = Skill(meta=SkillMeta(name='ux', description='Analyze bundles.'), body='steps')
>>> ad = complete(s)
>>> ad.name, ad.skills, ('Return contract' in ad.prompt)
('ux', ['ux'], True)

coact.diff(skill: object, agent: object) → AgentDiff

Show what extras agent adds over its source skill.

skill is any skill source; agent is an AgentDefinition, an agent *.md path, or a skill source (which is completed first).

>>> from skill.base import Skill, SkillMeta
>>> s = Skill(meta=SkillMeta(name='ux', description='Analyze bundles.'), body='steps')
>>> from coact import complete
>>> d = diff(s, complete(s))
>>> any(f == 'prompt' and cls.startswith('extra') for f, cls, _ in d.rows)
True

coact.emit_agent(ad: AgentDefinition, target: str = 'claude-agents-md', *, dest: str | Path | None = None) → Any

Emit an agent to a registered target; optionally write a file target to dest.

For string targets (e.g. claude-agents-md) with dest given, writes <dest>/<name>.md and returns the Path. Otherwise returns the emitter’s value (str or dict).

>>> ad = AgentDefinition(name='ux', description='Analyze.', prompt='You are...')
>>> emit_agent(ad).splitlines()[0]
'---'

coact.estimate(agents: object) → Estimate

Surface the fan-out cost tradeoff for an agent set (DECISIONS D9).

>>> from coact import AgentDefinition
>>> a = AgentDefinition(name='a', description='x', skills=['shared'])
>>> b = AgentDefinition(name='b', description='y', skills=['shared'])
>>> est = estimate([a, b])
>>> est.interdependent, est.shared_skills
(True, ['shared'])

coact.from_claude_agent_md(text: str) → AgentDefinition

Parse .claude/agents/*.md content back into an AgentDefinition.

The inverse of to_claude_agent_md(): lossless for every structured field; the prompt body is whitespace-trimmed on both ends (personas are not whitespace-sensitive).

>>> ad = AgentDefinition(name='ux', description='Analyze.', prompt='You are an analyst.',
...     tools=['Read', 'Grep'], model='sonnet',
...     returns=ReturnContract(json_schema={'type': 'object'}, description='out'))
>>> back = from_claude_agent_md(to_claude_agent_md(ad))
>>> (back.name, back.tools, back.model, back.prompt) == (ad.name, ad.tools, ad.model, ad.prompt)
True
>>> back.returns.json_schema
{'type': 'object'}

coact.integration_spec_from(source: IntegrationSpec | str | Path | Callable[[...], Any] | Iterable[Any], *, name: str | None = None, description: str = '', version: str = '0.1.0', author: str | None = None) → IntegrationSpec

Coerce a capability source into an IntegrationSpec.

Accepts (and flattens lists of): a prebuilt IntegrationSpec, a 'module:function' ref string, a live callable (its __module__:__qualname__ becomes the ref), or a skill source carrying a coact: mcp: block. The resulting name is kebab-cased so it is a safe, machine-readable bundle identifier.

>>> integration_spec_from(['os.path:basename'], name='paths').tools
['os.path:basename']
>>> integration_spec_from('os.path:basename', name='My Tools').name
'my-tools'

coact.integration_spec_from_description(description: str, *, llm: Any = None, model: str | None = None, name: str | None = None, version: str = '0.1.0', author: str | None = None, prompt_template: str | None = None, infer_tool_schemas: bool = True) → IntegrationSpec

Refine a natural-language description into a draft IntegrationSpec.

The opt-in LLM ingress (DECISIONS D10): generation is routed through aix (provider-agnostic) by default, via oa’s prompt-as-function machinery.

Parameters:

• description – What the integration should do, in plain language.

• llm – The LLM backend. None → aix.chat (the default, multi-provider); a callable(prompt, **kwargs) -> str is used as-is (handy for tests); a str is treated as a model name passed to aix.chat.

• model – Explicit model id (overrides a model-name llm); None lets the backend resolve its own configured default.

• name – Force the integration name (else the model proposes one). Kebab-cased.

• version – Spec version string.

• author – Optional author name recorded on the spec.

• prompt_template – Override the authoring prompt (e.g. one managed in pyrompt). Must contain a single {description} placeholder.

• infer_tool_schemas – When a proposed tool lacks an input_schema, infer one from its description via oa.infer_schema_from_verbal_description (aix-backed). Best-effort — failures leave the schema None.

Returns:

An IntegrationSpec whose tool_specs hold the proposed tools. Tools the description bound to existing code become runnable refs (spec.runnable_refs()); the rest are design drafts to bind later.

Example (offline, with an injected backend):

>>> reply = (
...     '{"name": "wx", "description": "weather",'
...     ' "tools": [{"name": "get_weather", "description": "lookup",'
...     ' "input_schema": {"type": "object", "properties": {"city":'
...     ' {"type": "string"}}}, "handler": null}]}'
... )
>>> spec = integration_spec_from_description("a weather tool", llm=lambda p, **k: reply)
>>> spec.name, [t.name for t in spec.tool_specs], spec.runnable_refs()
('wx', ['get_weather'], [])

coact.inventory(project: Path | str | None = None) → Inventory

Enumerate skills, derived agents, and MCP-exposed tools in a project.

>>> import tempfile
>>> inv = inventory(tempfile.mkdtemp())
>>> inv.skills, inv.agents
([], [])

coact.parse_coact_meta(source: str | Path | Skill | dict) → CoactMeta

Parse a CoactMeta from a SKILL.md path/text/Skill/frontmatter dict.

Accepts the several shapes coact callers have on hand:

• a Skill (uses its source_path to re-read the raw frontmatter, since Skill.meta drops the coact key);

• a filesystem path to a skill directory or a SKILL.md file;

• raw SKILL.md text;

• an already-parsed frontmatter dict.

>>> parse_coact_meta("---\nname: x\ncoact:\n  model: haiku\n---\nbody").model
'haiku'

coact.plan_completion(source: str | Path | Skill, *, policy: CompletionPolicy | None = None, llm: object = None) → AgentPlan

Plan the skill→agent completion, recording the provenance of every field.

Accepts a Skill, a path to a skill directory / SKILL.md, or a skill key/name resolvable in the local store or project skills. Pass an optional llm (any callable(str)->str, an aw StepConfig, or a model name) to draft a richer persona — the mechanical path needs none.

>>> from skill.base import Skill, SkillMeta
>>> s = Skill(meta=SkillMeta(name='auditor', description='Audit a bundle for issues.'), body='steps')
>>> plan = plan_completion(s)
>>> plan.agent.name
'auditor'
>>> plan.agent.model  # read-only description with default tools -> haiku
'haiku'
>>> any(p.field == 'prompt' for p in plan.provenance)
True

coact.publish(source: IntegrationSource, *, target: str = 'claude-local-mcpb', dry_run: bool = False, **kwargs: Any) → PublishResult

Publish a capability to a chatbot host via the named target.

source is anything coact.integration.integration_spec_from() accepts: 'module:function' refs, live callables, a skill carrying a coact: mcp: block, or a prebuilt IntegrationSpec. Target-specific options (dest, name, author, …) pass through as keyword arguments.

coact.publish_mcpb(source: IntegrationSpec | str | Path | Callable[[...], Any] | Iterable[Any], *, dest: str | None = None, dry_run: bool = False, name: str | None = None, author: str | None = None, version: str = '0.1.0', description: str = '', python_command: str = 'python', manifest_version: str = '0.3') → PublishResult

Bundle source into a Claude Desktop .mcpb extension.

>>> res = publish_mcpb(['os.path:basename'], name='paths', dry_run=True)
>>> res.dry_run, res.artifact, sorted(res.files)
(True, None, ['manifest.json', 'server/main.py', 'server/py2mcp_config.json'])

coact.publish_remote(source: IntegrationSpec | str | Path | Callable[[...], Any] | Iterable[Any], *, dest: str | None = None, dry_run: bool = False, name: str | None = None, author: str | None = None, version: str = '0.1.0', description: str = '', connector_url: str | None = None, idp_issuer: str | None = None, jwks_uri: str | None = None, audience: str | None = None, required_scopes: list | None = None, host: str = '127.0.0.1', port: int = 8000, transport: str = 'streamable-http', include_dockerfile: bool = True) → PublishResult

Scaffold a remote Claude connector (Streamable-HTTP MCP server + OAuth 2.1).

source is anything coact.integration.integration_spec_from() accepts. The OAuth parameters describe the managed IdP that issues tokens and this server’s public identity; when omitted, placeholder values + a loud warning are emitted so the scaffold is still useful (fill them in before going public).

>>> res = publish_remote(['os.path:basename'], name='paths', dry_run=True)
>>> res.dry_run, sorted(res.files)
(True, ['DEPLOY.md', 'Dockerfile', 'requirements.txt', 'server/app.py',
        'server/connector_config.json'])

coact.publish_targets() → list[str]

The names of all registered publish targets.

>>> isinstance(publish_targets(), list)
True

coact.realize(target: AgentDefinition | str | Path | Skill | list, *, backend: str = 'host', **kwargs) → Any

Realize an agent (or skill, or list) via the named backend.

>>> from coact import AgentDefinition
>>> import tempfile
>>> ad = AgentDefinition(name='ux', description='Analyze.', prompt='You are...', skills=['ux'])
>>> res = realize(ad, backend='host', dest=tempfile.mkdtemp(), link=False)
>>> res.agents['ux'].name
'ux.md'

coact.realize_crewai(target: AgentDefinition | str | Path | Skill | list, *, model_map: dict | None = None, default_model: str = 'openai/gpt-4o-mini', runner: Callable[[...], Any] | None = None, tools_map: dict | None = None, use_response_format: bool = True, policy: CompletionPolicy | None = None) → RunnableCrewAIAgent

Realize one agent as a CrewAI-backed RunnableCrewAIAgent (aw-compatible).

model_map overrides how coact selectors map to LiteLLM model strings; tools_map opt-in binds host-resolved tool names to callables/tools (D12); runner injects the run call for testing. Raises if asked to realize more than one agent (topology is out of scope — D8).

coact.realize_host(target: AgentDefinition | str | Path | Skill | list, *, scope: str = 'project', dest: Path | str | None = None, project_dir: Path | str | None = None, link: bool = True, skills_source: Path | str | list | None = None, force: bool = False, dry_run: bool = False, policy: CompletionPolicy | None = None) → RealizedHost

Materialize agents (+ linked skills) so the host agent runs them.

Writes <dest>/<name>.md for each agent and (when link) symlinks each referenced skill into the sibling .claude/skills/ so Claude Code discovers both. skills_source (a dir or list of dirs each holding <name>/SKILL.md) is searched first; otherwise skills are resolved by name via the local store / project. Verifies discovery and reports anything missing.

Pass dry_run=True to preview — the returned RealizedHost lists the agent files that would be written and the skills that would link (with the same unresolvable-skill warnings), but no file or symlink is created. This mirrors coact.complete.plan_completion()’s look-before-you-leap contract for the one backend that mutates the filesystem (progressive disclosure).

coact.realize_langgraph(target: AgentDefinition | str | Path | Skill | list, *, model_map: dict | None = None, default_model: str = 'openai:gpt-4o-mini', factory: Callable[[...], Any] | None = None, runner: Callable[[Any, Any], Any] | None = None, tools_map: dict | None = None, structured_strategy: str = 'auto', use_response_format: bool = True, policy: CompletionPolicy | None = None) → RunnableLLMGraphAgent

Realize one agent as a LangGraph-backed RunnableLLMGraphAgent (aw-compatible).

model_map overrides how coact selectors (sonnet/opus/haiku) map to langchain provider:model strings. tools_map opt-in binds host-resolved tool names to Python callables (D12). factory injects the graph builder for testing. Raises if asked to realize more than one agent (topology is out — D8).

coact.realize_litellm(target: AgentDefinition | str | Path | Skill | list, *, model_map: dict | None = None, default_model: str = 'openai/gpt-4o-mini', completion: Callable[[...], Any] | None = None, use_response_format: bool = True, policy: CompletionPolicy | None = None) → RunnableLLMAgent

Realize one agent as a provider-agnostic RunnableLLMAgent (via LiteLLM).

model_map overrides how coact model selectors (sonnet/opus/haiku) map to LiteLLM model strings — point them at any provider to prove portability.

coact.realize_mcp(target: AgentDefinition | str | Path | Skill | list, *, name: str | None = None, input_trans: Callable[[dict], dict] | None = None) → Any

Expose a skill’s declared Python tools as a FastMCP server (foreign-host).

Reads the coact: mcp: block (module + functions) of the source skill(s) and delegates to py2mcp.mk_mcp_from_refs — coact writes no MCP plumbing (DECISIONS §6.1.3). target may be a skill source or an AgentDefinition (whose source_skill is resolved back to the skill that carries the declaration).

coact.realize_sdk(target: AgentDefinition | str | Path | Skill | list, *, llm: Any = None, runner: Callable[[str, Any], Any] | None = None, return_mode: str = 'auto', policy: CompletionPolicy | None = None) → RunnableAgent

Realize a single agent as a runnable RunnableAgent (aw-compatible).

return_mode selects how the return contract reaches the SDK (DECISIONS D6): 'auto' uses native output_format when the installed SDK supports it and otherwise falls back to a forced return_result tool; pass 'tool' to force that fallback (e.g. for models / extended-thinking modes that cannot honor output_format).

coact.register_validator() → None

Register the coact-block validator into skill.create.validators (idempotent).

coact.resolve_llm(llm: Any = None) → Callable[[str], str] | None

Resolve llm to a callable(str) -> str, or None if unavailable.

>>> resolve_llm(lambda p: 'hi')('x')
'hi'
>>> resolve_llm('no-such-thing') is None or callable(resolve_llm('no-such-thing'))
True

coact.scaffold_fleet(target: AgentDefinition | str | Path | Skill | list, *, dest: str | Path | None = None, agents_dir: str = '.claude/agents', policy: CompletionPolicy | None = None) → str | Path

Emit a starter shim wiring the target agents as a runnable sdk fleet.

target is anything coact.realize() accepts (an AgentDefinition, a skill source, an agent *.md path, or a list of these). Returns the shim source string; if dest is given (a file path, or a directory in which fleet.py is written) the file is written and its Path returned instead.

The shim references each agent by name under agents_dir and chains them sequentially — a deliberately thin starting point. It is the only topology-adjacent artifact coact produces (DECISIONS D8): coact writes it once and never runs it; you own the control flow from there.

>>> from coact import AgentDefinition
>>> shim = scaffold_fleet([AgentDefinition(name='collector', description='x'),
...                         AgentDefinition(name='summarizer', description='y')])
>>> 'AGENTS = ["collector", "summarizer"]' in shim
True
>>> 'backend="sdk"' in shim and 'YOU OWN THIS FILE' in shim
True

coact.structured(prompt: str, schema: dict, *, llm: Any = None, retries: int = 1) → dict | None

Best-effort schema-conforming dict from an LLM, or None if unavailable.

Native structured output is provider-specific; this facade stays portable by instructing JSON-only output that conforms to schema and parsing it, retrying once on a parse miss. Returns None when no LLM is resolvable — callers fall back to a template (DECISIONS D10).

coact.synthesize_persona(skill: Skill, *, return_contract: ReturnContract, tools: list[str] | None = None, extra_skills: list[str] | None = None, llm: object = None) → tuple[str, Literal['skill', 'coact-frontmatter', 'policy', 'inferred', 'synthesized-template', 'synthesized-llm', 'default']]

Synthesize the system prompt (persona) for an agent derived from skill.

Wraps the skill’s intent as an identity, states operating invariants, and appends the return contract — while pointing at the source skill rather than inlining it (§3.3). The identity paragraph is template-generated by default; when an llm is resolvable it is drafted by the LLM instead (richer, but the invariants and the return contract stay deterministic so the machine-facing contract is never at the mercy of generation). The LLM is optional — absent one, the template path is a sound default (DECISIONS D10).

>>> from skill.base import Skill, SkillMeta
>>> s = Skill(meta=SkillMeta(name='ux-analyst', description='Analyze UX bundles.'), body='steps')
>>> rc = ReturnContract(json_schema={'type': 'object'}, description='findings')
>>> persona, src = synthesize_persona(s, return_contract=rc, tools=['Read', 'Grep'])
>>> 'ux-analyst' in persona and 'Return contract' in persona and src == 'synthesized-template'
True

coact.synthesize_return_contract(skill: Skill, *, coact_meta: CoactMeta) → tuple[ReturnContract, Literal['skill', 'coact-frontmatter', 'policy', 'inferred', 'synthesized-template', 'synthesized-llm', 'default']]

Choose the agent’s return contract: author-pinned, else a template default.

>>> from skill.base import Skill, SkillMeta
>>> s = Skill(meta=SkillMeta(name='x', description='y'), body='z')
>>> rc, src = synthesize_return_contract(s, coact_meta=CoactMeta())
>>> rc.json_schema['type'], src
('object', 'synthesized-template')
>>> rc2, src2 = synthesize_return_contract(s, coact_meta=CoactMeta(returns={'schema_ref': 'm:N'}))
>>> rc2.ref, src2
('m:N', 'coact-frontmatter')

coact.to_aw_agent_spec(ad: AgentDefinition)

Adapt an AgentDefinition to an aw.translators.AgentSpec.

Lets coact reuse aw’s CrewAI / OpenAI / SKILL.md renderers (the *_from_spec cores) instead of reimplementing them — the ecosystem coordination point. Requires aw.

coact.to_claude_agent_md(ad: AgentDefinition) → str

Serialize an agent to .claude/agents/<name>.md content.

>>> ad = AgentDefinition(name='ux', description='Analyze.', prompt='You are...')
>>> md = to_claude_agent_md(ad)
>>> md.startswith('---') and 'name: ux' in md and 'You are...' in md
True

coact.validate_coact_block(skill_or_meta: Skill | dict) → list[str]

Validate a coact: block, returning issue strings (empty = valid/absent).

Additive: a skill with no coact: block passes trivially. Checks key spelling, model/memory enums, mcp entry shape, and that returns carries exactly one of schema_ref/json_schema.

>>> validate_coact_block({'coact': {'model': 'gpt-4'}})
["coact.model: 'gpt-4' is not one of sonnet, opus, haiku, inherit"]
>>> validate_coact_block({'coact': {'mcp': [{'functions': ['f']}]}})
["coact.mcp[0]: missing required 'module'"]
>>> validate_coact_block({'name': 'ok'})
[]