Skip to content

Plugins

Structural protocols — implement the right methods and any class qualifies. No base class, no registration decorator.

See Plugins for a full authoring guide.

ModelProvider

civitas.plugins.model.ModelProvider

Bases: Protocol

Protocol for LLM invocation abstraction.

chat(model, messages, tools=None) async

Send a chat completion request to the model.

civitas.plugins.model.ModelResponse(content, model, tokens_in, tokens_out, cost_usd=None, tool_calls=None) dataclass

Response from an LLM call.

cost_usd = None class-attribute instance-attribute

Computed cost in USD, or None if the model's pricing is not known.

civitas.plugins.model.ToolCall(id, name, input) dataclass

A single tool call requested by the model.

Normalized across providers: Anthropic tool_use blocks and LiteLLM tool_calls entries both map to this shape.

Implementations

Provider implementations (Anthropic, OpenAI, Gemini, Mistral, LiteLLM) live in civitas-contrib. Install with e.g. pip install civitas-contrib[anthropic].

ToolProvider & ToolRegistry

civitas.plugins.tools.ToolProvider

Bases: Protocol

Protocol for external tool/API invocation with schema.

name property

Human-readable tool name used for lookup.

schema property

JSON Schema describing the tool's input parameters.

execute(**kwargs) async

Invoke the tool with the given keyword arguments.

civitas.plugins.tools.ToolRegistry()

Holds registered tools, injected into AgentProcess as self.tools.

Source code in civitas/plugins/tools.py 
29
30
def __init__(self) -> None:
    self._tools: dict[str, ToolProvider] = {}

register(tool)

Register a tool by its name.

Raises ValueError on duplicate name — silent overwrite would cause the wrong implementation to be called since names are used by the model.

Source code in civitas/plugins/tools.py 
32
33
34
35
36
37
38
39
40
41
42
def register(self, tool: ToolProvider) -> None:
    """Register a tool by its name.

    Raises ``ValueError`` on duplicate name — silent overwrite would cause
    the wrong implementation to be called since names are used by the model.
    """
    if tool.name in self._tools:
        raise ValueError(
            f"Tool already registered: {tool.name!r}. Deregister the existing tool first."
        )
    self._tools[tool.name] = tool

get(name)

Look up a tool by name.

Source code in civitas/plugins/tools.py 
54
55
56
def get(self, name: str) -> ToolProvider | None:
    """Look up a tool by name."""
    return self._tools.get(name)

StateStore

civitas.plugins.state.StateStore

Bases: Protocol

Protocol for agent state persistence.

get(agent_name) async

Retrieve the persisted state for an agent, or None if absent.

set(agent_name, state) async

Persist state for an agent.

delete(agent_name) async

Remove persisted state for an agent.

list_agents() async

Return all agent names with persisted state.

close() async

Release resources held by the store (connections, file handles).

civitas.plugins.state.InMemoryStateStore()

Default state store — in-memory, no persistence. State is lost on restart.

Source code in civitas/plugins/state.py 
36
37
def __init__(self) -> None:
    self._data: dict[str, dict[str, Any]] = {}

get(agent_name) async

Retrieve the in-memory state for an agent.

Source code in civitas/plugins/state.py 
39
40
41
async def get(self, agent_name: str) -> dict[str, Any] | None:
    """Retrieve the in-memory state for an agent."""
    return self._data.get(agent_name)

set(agent_name, state) async

Store state for an agent in memory (shallow copy to prevent aliasing).

Without copying, subsequent mutations to the caller's dict would also mutate the stored checkpoint, making restore-from-checkpoint a no-op.

Source code in civitas/plugins/state.py 
43
44
45
46
47
48
49
async def set(self, agent_name: str, state: dict[str, Any]) -> None:
    """Store state for an agent in memory (shallow copy to prevent aliasing).

    Without copying, subsequent mutations to the caller's dict would also
    mutate the stored checkpoint, making restore-from-checkpoint a no-op.
    """
    self._data[agent_name] = dict(state)

delete(agent_name) async

Remove state for an agent from memory.

Source code in civitas/plugins/state.py 
51
52
53
async def delete(self, agent_name: str) -> None:
    """Remove state for an agent from memory."""
    self._data.pop(agent_name, None)

list_agents() async

Return all agent names with persisted state, sorted.

Source code in civitas/plugins/state.py 
55
56
57
async def list_agents(self) -> list[str]:
    """Return all agent names with persisted state, sorted."""
    return sorted(self._data.keys())

close() async

No-op — in-memory store holds no external resources.

Source code in civitas/plugins/state.py 
59
60
async def close(self) -> None:
    """No-op — in-memory store holds no external resources."""

SQLiteStateStore and PostgresStateStore live in civitas-contrib. Install with pip install civitas-contrib or pip install civitas-contrib[postgres].

Plugin Loading

civitas.plugins.loader.load_plugins_from_config(config)

Load all plugins from a YAML configuration dict.

Expected YAML structure

plugins: models: - type: anthropic config: api_key: "sk-..." default_model: "claude-sonnet-4-20250514" exporters: - type: console - type: otel config: endpoint: "http://localhost:4317" state: type: in_memory

Returns a dict with keys: model_providers, exporters, state_store.

Source code in civitas/plugins/loader.py 
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
def load_plugins_from_config(config: dict[str, Any]) -> dict[str, Any]:
    """Load all plugins from a YAML configuration dict.

    Expected YAML structure:
        plugins:
          models:
            - type: anthropic
              config:
                api_key: "sk-..."
                default_model: "claude-sonnet-4-20250514"
          exporters:
            - type: console
            - type: otel
              config:
                endpoint: "http://localhost:4317"
          state:
            type: in_memory

    Returns a dict with keys: model_providers, exporters, state_store.
    """
    plugins_cfg = config.get("plugins", {})
    result: dict[str, Any] = {
        "model_providers": [],
        "exporters": [],
        "state_store": None,
    }

    # Models
    for model_cfg in plugins_cfg.get("models", []):
        name = model_cfg.get("type")
        if not name:
            raise PluginError(
                "model", "<missing>", "Plugin config entry is missing a 'type' field."
            )
        plugin_config = model_cfg.get("config", {})
        provider = load_plugin("model", name, plugin_config)
        result["model_providers"].append(provider)

    # Exporters
    for exp_cfg in plugins_cfg.get("exporters", []):
        name = exp_cfg.get("type")
        if not name:
            raise PluginError(
                "exporter", "<missing>", "Plugin config entry is missing a 'type' field."
            )
        plugin_config = exp_cfg.get("config", {})
        exporter = load_plugin("exporter", name, plugin_config)
        result["exporters"].append(exporter)

    # State store
    state_cfg = plugins_cfg.get("state")
    if state_cfg is not None:
        name = state_cfg.get("type", "in_memory")
        plugin_config = state_cfg.get("config", {})
        if name == "encrypted":
            result["state_store"] = _build_encrypted_store(plugin_config)
        else:
            result["state_store"] = load_plugin("state", name, plugin_config)

    return result