Utilities API Reference¶

Root-level modules: approval, cache, events, fallback, guardrails, streaming, tool optimization, adaptive strategy, context engine, env resolution, types, and exceptions.

Environment Variables¶

Parse and resolve ${ENV_VAR} syntax with default values and recursive resolution.

resolve_env_var¶

`promptise.env_resolver.resolve_env_var(value, *, context=None, allow_missing=False)` ¶

Resolve environment variables in a string.

Supports syntax: - ${VAR_NAME} - Required variable (raises if not found) - ${VAR_NAME:-default} - Optional variable with default

Parameters:

Name	Type	Description	Default
`value`	`str`	String potentially containing ${VAR} references.	required
`context`	`str \| None`	Optional context for error messages (e.g., "servers.math.url").	`None`
`allow_missing`	`bool`	If True, leave unresolved vars as-is instead of raising.	`False`

Returns:

Type	Description
`str`	String with all environment variables resolved.

Raises:

Type	Description
`EnvVarNotFoundError`	If a required variable is not found and allow_missing is False.

Examples:

>>> os.environ["API_KEY"] = "secret123"
>>> resolve_env_var("Bearer ${API_KEY}")
'Bearer secret123'
>>> resolve_env_var("${MISSING:-default}")
'default'
>>> resolve_env_var("${MISSING}")  # Raises EnvVarNotFoundError

resolve_env_in_dict¶

`promptise.env_resolver.resolve_env_in_dict(data, *, context_prefix='')` ¶

Recursively resolve environment variables in a dictionary.

Resolves ${VAR} references in all string values throughout a nested dictionary structure, including in lists and nested dicts.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Dictionary potentially containing env var references.	required
`context_prefix`	`str`	Prefix for error context (e.g., "servers.math").	`''`

Returns:

Type	Description
`dict[str, Any]`	New dictionary with all env vars resolved.

Raises:

Type	Description
`EnvVarNotFoundError`	If any required variable is not found.

Examples:

>>> os.environ["TOKEN"] = "abc123"
>>> resolve_env_in_dict({"auth": "${TOKEN}"})
{'auth': 'abc123'}
>>> resolve_env_in_dict({"server": {"url": "${API_URL}"}})
{'server': {'url': 'http://...'}}

validate_all_env_vars_available¶

`promptise.env_resolver.validate_all_env_vars_available(data)` ¶

Check which environment variables are referenced but not available.

This is useful for pre-validation before attempting to load a config. Scans the entire data structure for ${VAR} references and checks if each variable is set in the environment.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Parsed configuration data (dict from YAML/JSON).	required

Returns:

Type	Description
`list[str]`	List of missing environment variable names (empty if all available).
`list[str]`	Variables with defaults (${VAR:-default}) are not included.

Examples:

>>> data = {"api_key": "${OPENAI_API_KEY}", "url": "${API_URL:-http://default}"}
>>> missing = validate_all_env_vars_available(data)
>>> if missing:
...     print(f"Missing env vars: {', '.join(missing)}")

MCP Tool Discovery¶

ToolInfo¶

`promptise.tools.ToolInfo` `dataclass` ¶

Human-friendly metadata for a discovered MCP tool.

MCPClientError¶

`promptise.tools.MCPClientError` ¶

Bases: RuntimeError

Raised when communicating with the MCP client fails.

Approval (Human-in-the-loop)¶

Pause the agent before destructive actions and wait for explicit human approval.

ApprovalRequest¶

`promptise.approval.ApprovalRequest` `dataclass` ¶

A request for human approval of a tool call.

Attributes:

Name	Type	Description
`request_id`	`str`	Unique cryptographic ID for this request.
`tool_name`	`str`	Name of the tool requiring approval.
`arguments`	`dict[str, Any]`	Tool arguments (redacted if configured).
`agent_id`	`str \| None`	Agent or process identifier: the agent identity's id or `build_agent(observer_agent_id=...)`; in the Agent Runtime, the process name. `None` when the agent has neither.
`caller_user_id`	`str \| None`	User who triggered the agent (from CallerContext).
`context_summary`	`str`	The last few user/assistant messages of the conversation, for reviewer context (redacted like the arguments). See `ApprovalPolicy(context_messages=...)`.
`timestamp`	`float`	When the request was created (`time.time()`).
`timeout`	`float`	Seconds until auto-deny/allow.
`metadata`	`dict[str, Any]`	`source` (`"agent"` for the agent's own gate), `session_id` and `tenant_id` when known, plus whatever `ApprovalPolicy(metadata=...)` adds.
`tool_annotations`	`dict[str, Any]`	The tool's MCP annotations (`readOnlyHint`, `destructiveHint`, `idempotentHint`, `openWorldHint`, `title`) when it has them; empty otherwise. Hints from the server, not guarantees.
`raw_arguments`	`dict[str, Any] \| None`	The unredacted arguments, set by the agent's gate for in-process rule evaluation (:class:`~promptise.approval_classifier.AutoApprovalClassifier` matches its rules against them). Never serialized, signed or shown in `repr()`; handlers that display or log requests use `arguments`. `None` outside the agent's gate.

`compute_hmac(secret)` ¶

Compute the HMAC-SHA256 signature sent as X-Promptise-Signature.

Covers the fields in :data:SIGNED_FIELDS (request_id, tool_name, arguments, agent_id, caller_user_id, timestamp), serialized as sorted-key JSON. Verify it on the receiving side with :func:verify_webhook_signature.

`to_dict()` ¶

Serialize to a JSON-safe dict (for webhook payloads).

ApprovalDecision¶

`promptise.approval.ApprovalDecision` `dataclass` ¶

The decision on an approval request.

Attributes:

Name	Type	Description
`approved`	`bool`	Whether the tool call is approved.
`modified_arguments`	`dict[str, Any] \| None`	If the reviewer edited arguments.
`reviewer_id`	`str \| None`	Who made the decision.
`reason`	`str \| None`	Optional explanation.
`timestamp`	`float`	When the decision was made.
`decided_by`	`Literal['reviewer', 'classifier', 'gate']`	What made the decision: `"reviewer"` (the handler — a person, or your own handler code; the default), `"classifier"` (an :class:`~promptise.approval_classifier.AutoApprovalClassifier` rule, its read-only check or its LLM classifier) or `"gate"` (the agent's approval gate itself: `max_pending`, the `max_retries_after_deny` limit, a timeout or a handler error). Only `"reviewer"` denials (and timeouts) count towards `max_retries_after_deny`.
`trace`	`ClassifierDecisionTrace \| None`	The classifier layer that produced the decision, when an :class:`~promptise.approval_classifier.AutoApprovalClassifier` was involved; `None` otherwise.

ApprovalHandler¶

`promptise.approval.ApprovalHandler` ¶

Bases: Protocol

Protocol for approval handlers.

Implementations receive an :class:ApprovalRequest and must return an :class:ApprovalDecision. The handler is async — it can await webhooks, poll APIs, or wait on queues.

`request_approval(request)` `async` ¶

Ask for a decision on request and return it.

CallbackApprovalHandler¶

`promptise.approval.CallbackApprovalHandler` ¶

Approval handler that delegates to an async Python callable.

The simplest handler — pass any async def handler(request) -> decision function and it will be called for each approval request.

Parameters:

Name	Type	Description	Default
`callback`	`Callable[..., Any]`	Async callable that receives an :class:`ApprovalRequest` and returns an :class:`ApprovalDecision`.	required

`request_approval(request)` `async` ¶

Delegate to the user-provided callback.

WebhookApprovalHandler¶

`promptise.approval.WebhookApprovalHandler` ¶

Approval handler that POSTs to a webhook URL and polls for a decision.

Sends the :class:ApprovalRequest as a JSON POST to url. Then polls poll_url (or url + "/" + request_id) for the decision.

Parameters:

Name	Type	Description	Default
`url`	`str`	Webhook URL to POST the approval request to.	required
`secret`	`str \| None`	HMAC secret for signing requests. If not provided, a random secret is generated (ephemeral — not useful for cross-process verification).	`None`
`poll_url`	`str \| None`	URL to poll for the decision. Defaults to `{url}/{request_id}`.	`None`
`poll_interval`	`float`	Seconds between poll attempts.	`2.0`
`headers`	`dict[str, str] \| None`	Custom HTTP headers (e.g., auth tokens).	`None`
`http_client`	`Any \| None`	Pre-configured `httpx.AsyncClient` (proxy, mTLS, custom CA). Not closed by the handler.	`None`
`allow_private_networks`	`bool`	Allow `url` and `poll_url` to point at localhost, private, link-local or reserved addresses. Off by default (SSRF protection); turn it on when the approval service runs on your own network, e.g. `https://ops.internal/approvals`.	`False`

Unless allow_private_networks is set, the host is resolved again before the POST and before every poll, the request is refused if it resolves to a non-public address, and it is sent to the address that was checked (keeping the Host header and TLS server name), so a DNS answer that changes after construction (DNS rebinding) cannot redirect it to an internal service. Redirects are never followed.

Raises:

Type	Description
`ValueError`	`url` or `poll_url` targets a private or internal address and `allow_private_networks` is not set.

`request_approval(request)` `async` ¶

POST request to webhook, poll for decision.

QueueApprovalHandler¶

`promptise.approval.QueueApprovalHandler` ¶

Approval handler using async queues for in-process UIs.

For Gradio, Streamlit, or other in-process UIs where the human reviewer is in the same Python process. The UI reads from :attr:request_queue and writes decisions to the handler via :meth:submit_decision.

Example::

handler = QueueApprovalHandler()

# UI thread reads approval requests
request = await handler.request_queue.get()
# Show to user, collect decision
handler.submit_decision(request.request_id, ApprovalDecision(approved=True))

Attributes:

Name	Type	Description
`request_queue`	`Queue[ApprovalRequest]`	Queue of pending :class:`ApprovalRequest` objects.

`request_approval(request)` `async` ¶

Enqueue request and wait for decision from the UI.

`submit_decision(request_id, decision)` ¶

Submit a decision for a pending request.

Called by the UI after the human reviewer makes a choice.

Parameters:

Name	Type	Description	Default
`request_id`	`str`	The `request_id` from the :class:`ApprovalRequest`.	required
`decision`	`ApprovalDecision`	The reviewer's decision.	required

Raises:

Type	Description
`KeyError`	If no pending request with this ID exists.

ApprovalPolicy¶

`promptise.approval.ApprovalPolicy` ¶

Configuration for human-in-the-loop approval.

Defines which tools require approval, how to request it, and what happens on timeout or repeated denial.

Parameters:

Name	Type	Description	Default
`tools`	`list[str]`	Glob patterns for tool names that require approval. Examples: `["send_email"]`, `["delete_", "payment_"]`.	required
`handler`	`ApprovalHandler \| Callable[..., Any]`	An :class:`ApprovalHandler` implementation or an async callable `(ApprovalRequest) -> ApprovalDecision`.	required
`timeout`	`float`	Seconds to wait for a decision before applying `on_timeout`. Default: 300 (5 minutes).	`300.0`
`on_timeout`	`Literal['deny', 'allow']`	What to do when timeout expires. `"deny"` (default) rejects the tool call. `"allow"` permits it.	`'deny'`
`include_arguments`	`bool`	Include tool arguments in the approval request. Set to `False` to hide arguments from reviewers. Classifier rules still see them (`raw_arguments`).	`True`
`redact_sensitive`	`bool`	Run the arguments and `context_summary` through PII/credential detection before sending them to the reviewer. Each string value is redacted on its own, so the arguments keep their structure. Only the reviewer's copy is redacted: an :class:`~promptise.approval_classifier.AutoApprovalClassifier` matches its rules against the real arguments.	`True`
`max_pending`	`int`	Maximum concurrent pending approvals per agent. Additional tool calls are denied without asking the handler (`decided_by="gate"`).	`10`
`max_retries_after_deny`	`int \| None`	After this many reviewer denials of the same tool within `deny_window` (and the same `deny_scope`), later calls are denied without asking the handler (`decided_by="gate"`). Denials by classifier rules don't count. `None` disables the limit.	`3`
`deny_window`	`float \| None`	Seconds a denial counts towards `max_retries_after_deny`. Default: 600 (10 minutes). `None` keeps denials until the agent is rebuilt.	`600.0`
`deny_scope`	`Literal['session', 'user', 'agent']`	Whose denials count together: `"session"` (default; the same user and `chat()` session), `"user"` (the same `CallerContext` user, across sessions) or `"agent"` (every caller of the agent). Without a `CallerContext`, callers share one count.	`'session'`
`sequential`	`bool`	Ask the reviewer one call at a time. When the model requests several gated calls in one turn they run concurrently, so by default their approval requests arrive together; with `True` each waits until the previous one is decided.	`False`
`context_messages`	`int`	How many of the conversation's last user/assistant messages go into `context_summary`. `0` leaves it empty.	`3`
`metadata`	`dict[str, Any] \| Callable[[str, dict[str, Any]], Any] \| None`	Extra :attr:`ApprovalRequest.metadata` for every request: a dict, or a callable `(tool_name, arguments) -> dict` (sync or async) that receives the call's unredacted arguments.	`None`
`on_decision`	`Callable[[ApprovalRequest, ApprovalDecision], Any] \| None`	Called with `(request, decision)` for every decision the agent's gate reaches — the handler's, a classifier rule's, and the gate's own (`max_pending`, the retry limit, a timeout, a handler error) — before the tool runs, and for server-side approval gates answered through MCP elicitation (`request.metadata["source"] == "mcp_elicitation"`). Sync or async. Use it for the audit log: `decision.decided_by` and `decision.trace` say what decided. `request.arguments` is the redacted copy. Errors it raises are logged and ignored.	`None`

`record_decision(request, decision)` `async` ¶

Pass a decision to on_decision, logging (not raising) its errors.

`redact_arguments(arguments)` `async` ¶

Redact sensitive data from arguments before sending to reviewer.

Each string value — including strings nested in dicts and lists — is scanned on its own with the guardrails PII/credential detectors, and detected spans are replaced by labels such as [EMAIL]. Keys, numbers and the structure are kept: {"to": "dana@example.com", "amount": 18.5} becomes {"to": "[EMAIL]", "amount": 18.5}.

Returns the arguments unchanged when redact_sensitive is off. If the guardrails module cannot be imported or the scan fails, a warning is logged and the arguments are returned unredacted.

`request_metadata(tool_name, arguments)` `async` ¶

The developer-supplied part of :attr:ApprovalRequest.metadata.

`requires_approval(tool_name)` ¶

Check if a tool name matches any approval pattern.

Uses fnmatch glob matching — supports * and ? wildcards.

`summarize_context(messages)` `async` ¶

Build context_summary from a conversation's messages.

Takes the last context_messages user/assistant messages that have text, one per line as user: ... / assistant: ... (each cut to 500 characters), redacted like the arguments when redact_sensitive is on. System and tool messages are skipped.

approval_elicitation_callback¶

`promptise.approval.approval_elicitation_callback(handler, *, server_name=None, in_flight=None, timeout=None, event_notifier=None)` ¶

Build an MCP elicitation_callback that asks an approval handler.

Lets the human behind a Promptise client approve server-side approval gates — e.g. ApprovalGateMiddleware with ElicitationApprover on an MCPcast-generated server, which asks the calling client to confirm a gated tool call through MCP elicitation. build_agent(approval=...) installs one per server automatically; use this directly with :class:~promptise.mcp.client.MCPClient::

client = MCPClient(
    transport="stdio",
    command="python",
    args=["petstore-mcp/server.py"],
    elicitation_callback=approval_elicitation_callback(
        CallbackApprovalHandler(ask_human),
        server_name="petstore",
        in_flight=lambda: client.in_flight_calls,
    ),
)

Each request becomes an :class:ApprovalRequest: context_summary carries the server's message, metadata the message, the requested schema and the server name (source="mcp_elicitation"). When exactly one tool call is in flight on the connection, tool_name and arguments are that call's — the arguments this client actually sent — and the handler runs in the caller's context (so get_current_caller() works and caller_user_id is set). When no call or several calls are in flight, the request cannot be tied to one, so tool_name is "", arguments is empty and metadata["in_flight_tools"] lists the candidates; the reviewer decides from the server's message alone.

The decision maps back as follows, failing closed at every step:

approved → accept, with the form's decision field set to True (and reason filled in when the form has one);
denied → decline;
a timeout or a handler error → decline — never accept, even when the policy says on_timeout="allow": that setting governs the agent's own gate, not one the server asked a human to clear;
modified_arguments → decline (the server has already bound the arguments; executing the originals after a reviewer changed them would run something nobody approved);
a request that is not a confirmation — URL mode, or a form whose fields are not one boolean from :data:CONFIRMATION_FIELDS plus an optional reason — is declined without asking the handler.

Parameters:

Name	Type	Description	Default
`handler`	`ApprovalPolicy \| ApprovalHandler \| Callable[..., Any]`	An :class:`ApprovalPolicy` (its handler, `timeout`, `include_arguments` and `redact_sensitive` apply; its tool patterns do not — the server decided the call needs approval), an :class:`ApprovalHandler`, or a callable accepted by :class:`CallbackApprovalHandler`. An :class:`~promptise.approval_classifier.AutoApprovalClassifier` is bypassed in favour of its human `fallback`: its auto-allow rules were written for the agent's own gate, and must not clear a gate the server put in front of a human.	required
`server_name`	`str \| None`	Name shown to the reviewer and recorded in metadata.	`None`
`in_flight`	`Callable[[], Sequence[Any]] \| None`	Returns the connection's in-flight calls — pass `lambda: client.in_flight_calls`. Without it, requests are never tied to a tool call.	`None`
`timeout`	`float \| None`	Seconds to wait for the handler. Defaults to the policy's `timeout`, or 300.	`None`
`event_notifier`	`Any`	Optional :class:`~promptise.events.EventNotifier`; `approval.requested` / `approval.granted` / `approval.denied` are emitted with `source="mcp_elicitation"`.	`None`

Returns:

Type	Description
`Callable[[Any, Any], Awaitable[Any]]`	An async `(context, params) -> ElicitResult` callable for
`Callable[[Any, Any], Awaitable[Any]]`	`MCPClient(elicitation_callback=...)`.

Semantic Cache¶

Serves cached responses for semantically similar queries. In-memory or Redis backend with per-user/per-session/shared scope isolation.

SemanticCache¶

`promptise.cache.SemanticCache` ¶

Semantic cache for agent responses.

Caches LLM responses by query similarity using local or cloud embeddings. Reduces API costs by 30-50% for workloads with repetitive queries.

Security: Default scope is per_user — each user gets an isolated cache partition. No CallerContext = no caching. Cached responses always pass through output guardrails.

Parameters:

Name	Type	Description	Default
`backend`	`str`	`"memory"` (default) or `"redis"`.	`'memory'`
`redis_url`	`str \| None`	Redis connection URL (when backend is `"redis"`).	`None`
`embedding`	`EmbeddingProvider \| str \| None`	An :class:`EmbeddingProvider`, a model name string, or `None` for the default local model.	`None`
`similarity_threshold`	`float`	Minimum cosine similarity for a cache hit.	`0.92`
`default_ttl`	`int`	Default time-to-live in seconds.	`3600`
`scope`	`str`	Cache isolation: `"per_user"` (default), `"per_session"`, or `"shared"`.	`'per_user'`
`max_entries_per_user`	`int`	Max entries per scope partition.	`1000`
`max_total_entries`	`int`	Max entries across all scopes.	`100000`
`encrypt_values`	`bool`	Encrypt cached values at rest (Redis only).	`False`
`ttl_patterns`	`dict[str, int] \| None`	Regex → TTL overrides for time-sensitive queries.	`None`
`invalidate_on_write`	`bool`	Evict the caller's cached responses when the agent calls a write tool (see :meth:`is_write_tool`).	`True`
`cache_multi_turn`	`bool`	Cache requests that carry earlier turns. Off by default: a follow-up such as "What river runs through it?" means something different in every conversation, so only single-message requests are cached. When on, the earlier messages are hashed into the cache key, so a follow-up only hits for the same conversation history.	`False`
`cache_tool_turns`	`bool`	Cache answers for turns in which the agent called tools. Off by default, so a cached answer never stands in for a tool call. When on, a turn is cached only if every tool it called is read-only; a turn that called a write tool is never cached, because replaying it would skip the write.	`False`
`write_tools`	`Sequence[str] \| None`	Tool names (`*` wildcards allowed) always treated as writes, whatever their annotations say.	`None`
`read_only_tools`	`Sequence[str] \| None`	Tool names (`*` wildcards allowed) treated as read-only, for tools without MCP annotations (for example `extra_tools`) or with wrong ones. `write_tools` wins when a name matches both.	`None`
`shared_data_acknowledged`	`bool`	Required when scope is `"shared"`.	`False`

# One-liner
cache = SemanticCache()

# Full config
cache = SemanticCache(
    backend="redis",
    redis_url="redis://localhost:6379",
    similarity_threshold=0.92,
    scope="per_user",
    ttl_patterns={r"current|now|today": 60},
)

agent = await build_agent(..., cache=cache)

`allows_conversation(messages)` ¶

Whether a request with these input messages may use the cache.

A request carrying earlier turns (any user, assistant or tool message before the last one) is only cached when cache_multi_turn=True. System messages do not count.

`check(query_text, *, context_fingerprint='', caller=None, model_id=None, instruction_hash='')` `async` ¶

Check for a cached response.

Returns the cached entry if a semantically similar query was previously cached with the same context, model, and instructions. Returns None on cache miss.

Parameters:

Name	Type	Description	Default
`query_text`	`str`	The user's query.	required
`context_fingerprint`	`str`	Hash of memory + history context.	`''`
`caller`	`Any \| None`	:class:`CallerContext` for scope isolation.	`None`
`model_id`	`str \| None`	LLM model identifier.	`None`
`instruction_hash`	`str`	Hash of the system instructions.	`''`

`check_dependencies()` ¶

Raise :class:ImportError if this cache cannot run here.

build_agent(cache=...) calls this, so a missing numpy or sentence-transformers fails the build with an install hint instead of silently disabling the cache on every request. Only checks that the packages are installed; :meth:warmup loads the model.

`close()` `async` ¶

Close the cache backend (Redis connections, etc.).

Called automatically by agent.shutdown().

`invalidate_for_write(tool_name, caller=None)` `async` ¶

Evict caller's cached responses after a write.

The agent calls this when a write tool (see :meth:is_write_tool) finishes or fails, so answers computed before the write are not served after it. Responses still in flight from before the write are not stored either. Does nothing with invalidate_on_write=False.

`is_write_tool(name, annotations=None)` ¶

Whether calling this tool may change data.

Decided in this order: a write_tools match is a write; a read_only_tools match is not; otherwise the tool's MCP readOnlyHint annotation decides. A tool with no annotation counts as a write -- the MCP default for readOnlyHint is false.

`purge_user(user_id, *, tenant_id=None)` `async` ¶

Remove all cached entries for a user.

Use for GDPR right-to-erasure compliance.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	The user whose cache to purge.	required
`tenant_id`	`str \| None`	The user's tenant, when tenant-scoped callers were used. Must match the `CallerContext.tenant_id` the entries were stored under.	`None`

Note

This purges the per-user scope (user:<id>), which is the default and the GDPR-relevant one. A cache created with scope="per_session" keys entries by the user's session, so purge_user does not remove them; they expire with their TTL.

Returns:

Type	Description
`int`	Number of entries removed.

`stats()` `async` ¶

Get cache performance statistics.

`store(query_text, response_text, output, *, context_fingerprint='', caller=None, model_id=None, instruction_hash='', tools_used=None, write_generation=None)` `async` ¶

Store a response in the cache.

Parameters:

Name	Type	Description	Default
`query_text`	`str`	The user's query.	required
`response_text`	`str`	The extracted response text.	required
`output`	`Any`	The full LangGraph output dict.	required
`context_fingerprint`	`str`	Hash of memory + history context.	`''`
`caller`	`Any \| None`	:class:`CallerContext` for scope isolation.	`None`
`model_id`	`str \| None`	LLM model identifier.	`None`
`instruction_hash`	`str`	Hash of the system instructions.	`''`
`tools_used`	`list[str] \| None`	List of tool names called during this invocation.	`None`
`write_generation`	`int \| None`	:meth:`write_generation` read when the request started. If a write invalidated the scope since, the response may predate the write and is not stored.	`None`

`warmup()` ¶

Pre-load the embedding model.

Call at startup to avoid download/load latency on first cache check.

`write_generation(caller=None)` ¶

Number of write invalidations so far in caller's scope.

Pass the value read when a request starts to :meth:store; the entry is dropped if a write invalidated the scope in between.

CacheEntry¶

`promptise.cache.CacheEntry` `dataclass` ¶

A single cached response.

Attributes:

Name	Type	Description
`query_text`	`str`	The original user query.
`response_text`	`str`	The extracted response text.
`output`	`Any`	The output to replay. The agent stores `{"messages": [answer]}` (the final assistant message only) and, on a hit, returns it after the current request's own messages.
`embedding`	`list[float]`	The query embedding vector.
`scope_key`	`str`	Isolation scope (e.g. `"user:user-42"`).
`context_fingerprint`	`str`	Hash of memory + history + prompt context.
`model_id`	`str`	LLM model that generated this response.
`instruction_hash`	`str`	Hash of the system instructions.
`checksum`	`str`	SHA-256 of response_text for corruption detection.
`created_at`	`float`	Wall-clock (`time.time()`) timestamp of creation.
`ttl`	`int`	Time-to-live in seconds.
`metadata`	`dict[str, Any]`	Extra info (tools_used, token count, etc.).
`similarity`	`float \| None`	Cosine similarity between the stored query and the query that found this entry. Set on the entries a search returns; `None` on stored entries.

`age` `property` ¶

Seconds since this entry was stored.

`expired` `property` ¶

Check if this entry has expired.

Uses time.time() (wall clock) consistently across all backends. Both InMemory and Redis backends set created_at with time.time().

`verify_checksum()` ¶

Verify response integrity.

CacheStats¶

`promptise.cache.CacheStats` `dataclass` ¶

Cache performance statistics.

Attributes:

Name	Type	Description
`hits`	`int`	Number of cache hits.
`misses`	`int`	Number of cache misses.
`stores`	`int`	Number of entries stored.
`evictions`	`int`	Number of entries evicted.
`hit_rate`	`float`	Proportion of requests served from cache.

EmbeddingProvider¶

`promptise.cache.EmbeddingProvider` ¶

Bases: Protocol

Protocol for embedding providers.

Implement this to plug in any embedding model or API::

class MyProvider:
    async def embed(self, texts: list[str]) -> list[list[float]]:
        return my_model.encode(texts)

LocalEmbeddingProvider¶

`promptise.cache.LocalEmbeddingProvider` ¶

Local embedding via sentence-transformers.

Uses the same model loading pattern as tool optimization. If the same model is already loaded (e.g. for semantic tool selection), the instance is shared — no duplicate memory.

Parameters:

Name	Type	Description	Default
`model`	`str`	Model name or local directory path.	`DEFAULT_MODEL`

provider = LocalEmbeddingProvider()  # all-MiniLM-L6-v2
provider = LocalEmbeddingProvider(model="BAAI/bge-small-en-v1.5")
provider = LocalEmbeddingProvider(model="/models/local/embeddings")

`embed(texts)` `async` ¶

Embed texts using the local model.

`warmup()` ¶

Pre-load the embedding model.

OpenAIEmbeddingProvider¶

`promptise.cache.OpenAIEmbeddingProvider` ¶

Embedding via OpenAI or Azure OpenAI API.

Parameters:

Name	Type	Description	Default
`model`	`str`	Model name (e.g. `"text-embedding-3-small"`).	`'text-embedding-3-small'`
`api_key`	`str \| None`	OpenAI API key.	`None`
`base_url`	`str \| None`	Custom base URL (for Azure or proxies).	`None`
`azure_endpoint`	`str \| None`	Azure OpenAI endpoint URL.	`None`
`azure_deployment`	`str \| None`	Azure deployment name.	`None`

# OpenAI
provider = OpenAIEmbeddingProvider(
    model="text-embedding-3-small",
    api_key="${OPENAI_API_KEY}",
)

# Azure OpenAI
provider = OpenAIEmbeddingProvider(
    model="text-embedding-3-small",
    azure_endpoint="https://xxx.openai.azure.com",
    azure_deployment="my-embedding",
    api_key="${AZURE_OPENAI_KEY}",
)

`embed(texts)` `async` ¶

Embed texts via OpenAI API.

InMemoryCacheBackend¶

`promptise.cache.InMemoryCacheBackend` ¶

In-memory cache with numpy-based similarity search.

Stores entries per scope with LRU eviction. Thread-safe via asyncio (single event loop). No persistence — lost on restart.

Parameters:

Name	Type	Description	Default
`max_entries_per_scope`	`int`	Max entries per scope partition.	`1000`
`max_total_entries`	`int`	Max entries across all scopes.	`100000`

`invalidate(scope_key, pattern=None)` `async` ¶

Evict entries matching a pattern (or all for scope).

`purge_user(user_id)` `async` ¶

Remove all entries for a user (GDPR compliance).

`search(scope_key, embedding, threshold, *, match=None)` `async` ¶

Find the most similar entry above threshold.

Only entries for which match returns True are considered, so a closer entry stored under another context, model or instruction set does not hide the one that applies. The returned entry is a copy with :attr:CacheEntry.similarity set. Hits and misses are counted by :class:SemanticCache, which knows whether the request was served.

`store(scope_key, entry)` `async` ¶

Store an entry, evicting LRU if at capacity.

RedisCacheBackend¶

`promptise.cache.RedisCacheBackend` ¶

Redis-backed cache with vector similarity search.

Stores cache entries as JSON in Redis hashes. Embeddings are stored per scope and similarity is computed by fetching all embeddings for a scope and running numpy dot product locally. This avoids requiring the RediSearch module while keeping similarity search functional.

Optional AES encryption at rest via encrypt_values=True. Encryption key is read from PROMPTISE_CACHE_KEY env var or auto-generated per process.

Parameters:

Name	Type	Description	Default
`redis_url`	`str`	Redis connection URL (e.g. `redis://localhost:6379`).	`'redis://localhost:6379'`
`max_entries_per_scope`	`int`	Max entries per scope partition.	`1000`
`max_total_entries`	`int`	Max entries across all scopes.	`100000`
`encrypt_values`	`bool`	Encrypt cached response values at rest.	`False`

`close()` `async` ¶

Close the Redis connection.

`invalidate(scope_key, pattern=None)` `async` ¶

Evict entries for a scope.

`purge_user(user_id)` `async` ¶

Remove all entries for a user (GDPR compliance).

`search(scope_key, embedding, threshold, *, match=None)` `async` ¶

Find the most similar entry above threshold.

Same contract as :meth:InMemoryCacheBackend.search: candidates are tried from most to least similar and the first one that is live, intact and accepted by match is returned, with :attr:CacheEntry.similarity set.

`store(scope_key, entry)` `async` ¶

Store an entry in Redis.

Event Notifications¶

Webhook + callback sinks for structured agent events (invocation errors, tool failures, budget violations, etc).

AgentEvent¶

`promptise.events.AgentEvent` `dataclass` ¶

A structured notification event from the Promptise framework.

Attributes:

Name	Type	Description
`event_type`	`str`	Dotted event name (e.g. `"invocation.complete"`).
`severity`	`str`	One of `"info"`, `"warning"`, `"error"`, `"critical"`.
`timestamp`	`float`	When the event occurred (`time.time()`).
`agent_id`	`str \| None`	Agent or process identifier.
`user_id`	`str \| None`	User who triggered the action (from CallerContext).
`session_id`	`str \| None`	Conversation session ID if applicable.
`data`	`dict[str, Any]`	Event-specific payload (tool name, error message, etc.).
`metadata`	`dict[str, Any]`	Agent configuration, model ID, etc.

`compute_hmac(secret)` ¶

Compute an HMAC-SHA256 over this event's sorted-key JSON.

This is not what :class:WebhookSink sends: webhook deliveries are signed over the exact body bytes plus a timestamp. Verify those with :func:verify_event_signature.

`to_dict()` ¶

Serialize to a JSON-safe dict.

EventSink¶

`promptise.events.EventSink` ¶

Bases: Protocol

Protocol for event notification sinks.

Sinks receive events from the :class:EventNotifier and deliver them to external systems (webhooks, logs, callbacks, etc.).

`emit(event)` `async` ¶

Deliver a single event.

EventNotifier¶

`promptise.events.EventNotifier` ¶

Central event coordinator that routes events to configured sinks.

Each sink gets its own queue and background delivery task, so a slow or retrying sink (a webhook whose receiver is down) never delays the other sinks, and a failing sink never affects them. Within one sink, events are delivered in order. The agent never blocks waiting for delivery.

Parameters:

Name	Type	Description	Default
`sinks`	`list[EventSink]`	List of :class:`EventSink` implementations.	required
`max_queue_size`	`int`	Maximum undelivered events per sink. When a sink's queue is full, new events are dropped for that sink and counted in :attr:`dropped_count`, with a warning log.	`1000`
`shutdown_timeout`	`float`	Seconds :meth:`stop` waits for queued events to be delivered before giving up. Events still undelivered then are dropped and logged (per sink, with their types).	`10.0`
`slow_tool_threshold`	`float \| None`	Seconds after which a tool call emits a `tool.slow` event. `None` disables `tool.slow`.	`5.0`

notifier = EventNotifier(sinks=[
    WebhookSink("https://hooks.slack.com/...", events=["invocation.error"]),
    CallbackSink(my_handler),
])
await notifier.start()
notifier.emit_sync(AgentEvent(event_type="invocation.start", severity="info"))
await notifier.stop()

`is_running` `property` ¶

Whether delivery tasks are running.

`sinks` `property` ¶

The configured sinks.

`add_sink(sink)` ¶

Add sink; from now on it receives every event too.

Idempotent (a sink already added is not added twice). When the notifier is running, the new sink's delivery task starts at once on the notifier's loop. Events emitted before the call are not replayed to it.

`emit(event)` `async` ¶

Queue an event for delivery (non-blocking, starts the notifier).

If a sink's queue is full, the event is dropped for that sink with a warning log.

`emit_sync(event)` ¶

Queue an event from a synchronous context.

Safe to call from any thread: from outside the notifier's event loop the event is handed over thread-safely. Inside a running loop, the notifier starts itself if needed. Never raises.

`flush(timeout=None)` `async` ¶

Wait until every queued event has been delivered.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Maximum seconds to wait (`None` = no limit).	`None`

Returns:

Type	Description
`bool`	`True` when everything was delivered, `False` on timeout
`bool`	(or when the notifier is not running and events are queued).

`start()` `async` ¶

Start the background delivery tasks (idempotent).

`stop(timeout=None)` `async` ¶

Deliver what is queued, then stop the delivery tasks.

Waits up to timeout seconds (default: shutdown_timeout). Events that are still undelivered after that are dropped, counted in :attr:dropped_count and logged with their sink and types. Sinks with a close() method (like :class:WebhookSink) are closed afterwards. The notifier can be started again.

WebhookSink¶

`promptise.events.WebhookSink` ¶

Deliver events via HTTP POST to a webhook URL.

Features: HMAC-SHA256 signing with a timestamp (replay protection), retry with exponential backoff, SSRF protection, per-event filtering, payload redaction.

Every request carries:

X-Promptise-Signature: t=<unix seconds>,v1=<hex> — HMAC-SHA256 over b"<t>." + body with secret. Check it with :func:verify_event_signature against the raw body.
X-Promptise-Timestamp — the same t, for convenience.
X-Promptise-Event — the event type.
X-Promptise-Delivery — an id that stays the same across retries of one event, so receivers can de-duplicate.

Parameters:

Name	Type	Description	Default
`url`	`str`	Webhook URL to POST events to.	required
`events`	`list[str] \| None`	Event types to subscribe to (`None` = all events).	`None`
`headers`	`dict[str, str] \| None`	Custom HTTP headers (e.g. auth tokens).	`None`
`secret`	`str \| None`	HMAC secret for signing payloads. If not provided, a random secret is generated (readable as :attr:`secret`).	`None`
`max_retries`	`int`	Maximum retry attempts on failure.	`3`
`retry_delay`	`float`	Initial retry delay in seconds (doubles each retry).	`1.0`
`redact_sensitive`	`bool`	Scan the whole payload (`data`, `user_id`, `session_id`, `metadata` …) for PII/credentials before sending.	`True`
`min_severity`	`str \| None`	Minimum severity level to emit.	`None`
`transform`	`Callable[[dict[str, Any]], dict[str, Any]] \| None`	Reshape the (redacted) payload before it is sent, e.g. into a Slack or PagerDuty message.	`None`
`allow_private_networks`	`bool`	Allow `localhost`, loopback and private IP ranges. Off by default as SSRF protection; turn it on for a receiver on your own machine or private network.	`False`

`secret` `property` ¶

The signing secret (generated when none was passed).

`close()` `async` ¶

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Arbitrary dict (event payload, observability metadata, etc.).	required

Returns:

Type	Description
`dict[str, Any]`	A new dict with sensitive values replaced by placeholders.

Model Fallback¶

Automatic provider failover. Uses a circuit breaker to route around unhealthy providers without adding latency.

FallbackChain¶

`promptise.fallback.FallbackChain` ¶

Bases: BaseChatModel

Chain of LLM models with automatic failover.

Tries models in order. If one fails (exception, timeout), the next is tried. Each model has an independent circuit breaker — after failure_threshold consecutive failures, the model is skipped for recovery_timeout seconds before being tested again.

Passes through to build_agent(model=...) seamlessly — it's a BaseChatModel subclass, so LangChain treats it like any other model. :meth:bind_tools binds tools to every model in the chain.

Each answer records the model that wrote it in AIMessage.response_metadata["fallback_model"].

Every exception moves on to the next model, but a request the provider rejects as invalid (HTTP 400/413/422, see :func:is_request_error) does not count toward the circuit breaker. Streaming falls back only until the first chunk: after that the answer is committed to the model that sent it, and a failure is raised.

Parameters:

Name	Type	Description	Default
`models`	`Sequence[str \| Model \| BaseChatModel] \| None`	Ordered list of model identifiers (strings like `"openai:gpt-5-mini"`) or `BaseChatModel` instances. First model is primary, rest are fallbacks.	`None`
`timeout_per_model`	`float`	Maximum seconds per model attempt. `0` = no per-model timeout (use provider default).	`0`
`global_timeout`	`float`	Maximum seconds across ALL attempts combined. `0` = unlimited (each model gets its full timeout).	`0`
`failure_threshold`	`int`	Consecutive failures before a model's circuit breaker opens. Default: 3.	`3`
`recovery_timeout`	`float`	Seconds before a tripped circuit breaker allows a test request. Default: 60.	`60.0`
`on_fallback`	`Any`	Optional callback `(primary_model, fallback_model, error)` called each time a fallback is activated.	`None`

Raises:

Type	Description
`ValueError`	If `models` is empty.
`RuntimeError`	If all models in the chain fail.

`active_model` `property` ¶

Return the first non-skipped model's name.

`model_name` `property` ¶

Return the model that last served a request.

Before any request is made, returns the primary model's name. After a request, returns the model that actually served it — this is what observability and cache use.

`bind_tools(tools, *, tool_choice=None, **kwargs)` ¶

Bind tools to every model in the chain.

Returns a copy of the chain whose models all have the tools bound. The copy shares this chain's circuit breakers, so failures seen while calling tools count toward the same thresholds, and :attr:model_name / :meth:get_chain_status on this chain reflect requests served through the copy.

Raises:

Type	Description
`NotImplementedError`	If a model in the chain cannot call tools.

`get_chain_status()` ¶

Get the health status of each model in the chain.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts with `model_id`, `state` (closed/open/half_open),
`list[dict[str, Any]]`	`failures`, and `is_primary`.

Security Guardrails¶

Multi-head security scanner: prompt injection (DeBERTa ML), PII detection (69 regex patterns), credential detection (96 patterns), NER (GLiNER), content safety, custom rules.

PromptiseSecurityScanner¶

`promptise.guardrails.PromptiseSecurityScanner` ¶

Unified security scanner for agent input and output.

Compose detection heads to build exactly the scanner you need. Each head is a standalone config object — plug in what matters, leave out what doesn't.

Composable API (recommended)::

from promptise.guardrails import (
    PromptiseSecurityScanner,
    InjectionDetector,
    PIIDetector,
    CredentialDetector,
    ContentSafetyDetector,
    NERDetector,
    CustomRule,
)

scanner = PromptiseSecurityScanner(
    detectors=[
        InjectionDetector(),
        PIIDetector(categories={PIICategory.CREDIT_CARDS, PIICategory.SSN}),
        CredentialDetector(categories={CredentialCategory.AWS}),
    ],
    custom_rules=[
        CustomRule(name="internal_id", pattern=r"INT-\d{8}"),
    ],
)

One-liner defaults (all heads enabled)::

scanner = PromptiseSecurityScanner.default()

Flat API (backward compatible)::

scanner = PromptiseSecurityScanner(
    detect_injection=True,
    detect_pii={PIICategory.CREDIT_CARDS, PIICategory.SSN},
    detect_credentials={CredentialCategory.AWS},
)

Parameters:

Name	Type	Description	Default
`detectors`	`list[Any] \| None`	List of detector instances to enable.	`None`
`custom_rules`	`list[CustomRule \| dict[str, Any]] \| None`	List of :class:`CustomRule` instances.	`None`
`fail_open`	`bool`	What to do when an enabled head cannot run (its library is missing, its model fails to load, Ollama or Azure is unreachable). `False` (the default) fails closed: the scan gets a `BLOCK` finding, so `check_input` / `check_output` raise :class:`GuardrailViolation`. `True` logs a warning and lets the text through. Either way the head is reported in :attr:`ScanReport.scanners_skipped`, not in `scanners_run`.	`False`
`redact_input`	`bool`	Apply the PII and credential actions to input as well as output, and have :meth:`check_input` return the redacted text so the agent sends that to the model. Off by default: on input, PII and credentials are warnings and the message is passed unchanged.	`False`
`scan_tool_results`	`bool`	Scan every tool result before the model sees it (indirect prompt injection, leaked secrets). `build_agent` wraps the agent's tools when this is set; see :meth:`check_tool_result`.	`False`
`detect_injection`	`bool`	(flat API) Enable injection detection.	`True`
`detect_pii`	`bool \| set[PIICategory]`	(flat API) Enable PII detection.	`True`
`detect_toxicity`	`bool`	(flat API) Enable toxicity detection.	`True`
`detect_credentials`	`bool \| set[CredentialCategory]`	(flat API) Enable credential detection.	`True`

`check_input(text)` `async` ¶

Scan input text. Raises :class:GuardrailViolation on block.

Called by the agent before any processing (memory, tools, LLM).

Returns:

Type	Description
`str`	The text to send on: the redacted text when `redact_input`
`str`	is set and something was redacted, otherwise text unchanged.
`str`	The agent replaces the user's message with it.

`check_output(output)` `async` ¶

Scan output text. Redacts PII/credentials. Blocks on injection.

Called by the agent after the LLM response, before returning.

`check_tool_result(tool_name, result)` `async` ¶

Scan a tool result before the model reads it.

A tool result is where indirect prompt injection arrives (a web page, an email, a ticket), so every enabled head runs, the injection model included. PII and credentials get their configured actions, as on output, so a secret a tool returns is redacted before the model (or result["messages"]) holds it.

build_agent calls this for every tool call when scan_tool_results=True; a blocked result is replaced with a short notice the model can see, and the raw result is dropped.

Raises:

Type	Description
`GuardrailViolation`	With `direction="tool"` on a block.

`default(**kwargs)` `classmethod` ¶

Create a scanner with all detection heads enabled (defaults).

Equivalent to::

PromptiseSecurityScanner(detectors=[
    InjectionDetector(),
    PIIDetector(),
    CredentialDetector(),
])

Keyword arguments (fail_open, redact_input, scan_tool_results, custom_rules) are passed through.

`list_credential_patterns()` `staticmethod` ¶

Return names of all built-in credential patterns.

`list_pii_patterns()` `staticmethod` ¶

Return names of all built-in PII patterns.

`scan_text(text, *, direction='input')` `async` ¶

Run all enabled detection heads on the given text.

Parameters:

Name	Type	Description	Default
`text`	`str`	Text to scan.	required
`direction`	`str`	`"input"`, `"output"` or `"tool"` (a tool result). On input, PII and credential findings are warnings unless `redact_input` is set; on output and tool results they get their configured action. The injection head skips output.	`'input'`

Returns:

Name	Type	Description
`A`	`ScanReport`
	`ScanReport`	that could not run is listed in `scanners_skipped` and,
	`ScanReport`	unless `fail_open` is set, adds a `BLOCK` finding.

`warmup()` ¶

Pre-load ML models so the first scan is fast.

Call this at startup to avoid download/load latency on the first message, and to fail fast: a missing library or model, or an unreachable Ollama for :class:ContentSafetyDetector, raises here (whatever fail_open says) instead of surfacing on the first request. Safe to call multiple times (models are cached).

Example::

scanner = PromptiseSecurityScanner()
scanner.warmup()  # downloads + loads models NOW
agent = await build_agent(..., guardrails=scanner)

SecurityFinding¶

`promptise.guardrails.SecurityFinding` `dataclass` ¶

A single detection result from a scanner.

Attributes:

Name	Type	Description
`detector`	`str`	Which detection head found this (injection/pii/toxicity/credential).
`category`	`str`	Specific sub-category (e.g. `"credit_card_visa"`, `"aws_access_key"`).
`severity`	`Severity`	How severe this finding is.
`confidence`	`float`	Model confidence or 1.0 for regex matches.
`matched_text`	`str`	The text span that matched.
`start`	`int`	Character offset in original text.
`end`	`int`	Character offset in original text.
`action`	`Action`	What should happen (block/redact/warn).
`description`	`str`	Human-readable explanation.
`metadata`	`dict[str, Any]`	Extra information (model scores, pattern name, etc.).

ScanReport¶

`promptise.guardrails.ScanReport` `dataclass` ¶

Complete scan result with all findings and metadata.

Attributes:

Name	Type	Description
`passed`	`bool`	True if no findings have action=BLOCK.
`findings`	`list[SecurityFinding]`	All detections from all scanners.
`duration_ms`	`float`	Total scan time in milliseconds.
`scanners_run`	`list[str]`	Which detection heads ran.
`text_length`	`int`	Length of scanned text.
`redacted_text`	`str \| None`	Text with PII/credentials replaced (output scans).
`user_id`	`str \| None`	Owning user (from :class:`~promptise.agent.CallerContext`) for audit attribution. `None` when the scan ran outside of an invocation context.
`session_id`	`str \| None`	Conversation session attached to this scan, when available (from `caller.metadata['session_id']`).
`caller_roles`	`tuple[str, ...]`	The caller's roles at scan time, captured so that downstream audit log entries can rationalize the decision even after the caller context has moved on.
`scanners_skipped`	`dict[str, str]`	Detection heads that were enabled but could not run, mapped to the reason (`transformers` missing, model failed to load, Ollama unreachable, ...). A skipped head is never listed in `scanners_run`. With `fail_open=False` (the default) each skipped head also adds a `BLOCK` finding, so the scan fails.

`blocked` `property` ¶

Findings that caused a block.

`redacted` `property` ¶

Findings that were redacted.

`warnings` `property` ¶

Findings that are warnings only.

PIICategory¶

`promptise.guardrails.PIICategory` ¶

Bases: str, Enum

PII detection categories. Pass a set of these to PromptiseSecurityScanner(enable_pii={...}) to enable only specific PII types.

Example::

scanner = PromptiseSecurityScanner(
    enable_pii={PIICategory.CREDIT_CARDS, PIICategory.SSN, PIICategory.EMAIL},
)

CredentialCategory¶

`promptise.guardrails.CredentialCategory` ¶

Bases: str, Enum

Credential detection categories. Pass a set of these to PromptiseSecurityScanner(enable_credentials={...}).

Example::

scanner = PromptiseSecurityScanner(
    enable_credentials={
        CredentialCategory.AWS,
        CredentialCategory.OPENAI,
        CredentialCategory.GITHUB,
    },
)

Severity¶

`promptise.guardrails.Severity` ¶

Bases: str, Enum

Severity level of a security finding.

Action¶

`promptise.guardrails.Action` ¶

Bases: str, Enum

Action to take when a finding is detected.

Streaming¶

StreamEvent¶

`promptise.streaming.StreamEvent` `dataclass` ¶

Base event yielded by astream_with_tools().

All events have a type string and a timestamp. Use to_dict() for JSON serialization or to_json() for SSE.

Attributes:

Name	Type	Description
`type`	`str`	Event type identifier.
`timestamp`	`float`	Monotonic time when the event was created.

`to_dict()` ¶

Serialize to a JSON-safe dict.

`to_json()` ¶

Attributes:

Name	Type	Description
`message`	`str`	Human-readable error description.
`recoverable`	`bool`	Whether the agent might retry.

Stream helpers¶

`promptise.streaming.tool_display_name(tool_name, overrides=None)` ¶

Convert a tool name to a human-readable display string.

Examples:

"search_customers" → "Searching customers" "get_order_status" → "Getting order status" "hr_list_employees" → "Listing employees"

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	Raw tool name (underscore-separated).	required
`overrides`	`dict[str, str] \| None`	Developer-provided display name overrides.	`None`

Returns:

Type	Description
`str`	Human-readable display string.

`promptise.streaming.tool_summary(result, max_length=120)` ¶

Summarize a tool result for display.

JSON dicts: shows first 3 key-value pairs
JSON lists: shows item count
Plain text: truncates to max_length

Parameters:

Name	Type	Description	Default
`result`	`str \| None`	Raw tool result string.	required
`max_length`	`int`	Maximum summary length.	`120`

Returns:

Type	Description
`str`	One-line summary string.

`promptise.streaming.tool_error_summary(output, max_length=120)` ¶

Summarize a failed tool call's output for display.

An MCP ToolError arrives as {"error": {"code", "message", ...}}; its message is the summary. Any other output is summarized like :func:tool_summary.

Examples:

'{"error": {"code": "TOOL_ERROR", "message": "No order A-9."}}' → "No order A-9."

Parameters:

Name	Type	Description	Default
`output`	`str \| None`	Raw tool output.	required
`max_length`	`int`	Maximum summary length.	`120`

Returns:

Type	Description
`str`	One-line summary string.

`promptise.streaming.content_text(content)` ¶

The text in a message's (or chunk's) content.

content is a string, or a list of content blocks (Anthropic, Gemini…) of which only the "text" blocks are text.

Parameters:

Name	Type	Description	Default
`content`	`Any`	A `BaseMessage.content` value.	required

Returns:

Type	Description
`str`	The concatenated text, `""` when there is none.

Tool Optimization¶

Static schema minification + semantic tool selection (local embeddings) that offers each model call only the relevant tools. See Tool Optimization for measured savings.

OptimizationLevel¶

`promptise.tool_optimization.OptimizationLevel` ¶

Bases: str, Enum

Preset optimization levels.

MINIMAL — schema minification + description truncation.
STANDARD — deeper minification + nested description stripping.
SEMANTIC — all static optimizations + per-invocation semantic tool selection.

ToolOptimizationConfig¶

`promptise.tool_optimization.ToolOptimizationConfig` `dataclass` ¶

Configuration for MCP tool token optimization.

Pass a preset :attr:level for sensible defaults, or override individual settings for fine-grained control. Any field set explicitly takes precedence over the preset.

Parameters:

Name	Type	Description	Default
`level`	`OptimizationLevel \| None`	Preset optimization level.	`None`
`minify_schema`	`bool \| None`	Strip `description` from Pydantic Field metadata.	`None`
`max_description_length`	`int \| None`	Truncate tool descriptions at N chars.	`None`
`strip_nested_descriptions`	`bool \| None`	Remove descriptions from nested model fields (keeps top-level field descriptions).	`None`
`max_schema_depth`	`int \| None`	Flatten nested objects beyond this depth to `dict`. `None` means no limit.	`None`
`semantic_selection`	`bool \| None`	Enable semantic tool selection: before every model call, offer only the tools most relevant to the recent conversation. Requires `sentence-transformers` (`pip install "promptise[tool-optimization]"`).	`None`
`semantic_top_k`	`int \| None`	Number of most-relevant tools to offer per model call. Preserved tools, tools called in the previous or current turn, tools unlocked via `request_more_tools` and the fallback itself are added on top, up to 100 indexed tools per model call (:data:`MAX_OFFERED_TOOLS`).	`None`
`semantic_context_turns`	`int \| None`	How many of the most recent user turns make up the selection query (together with the latest assistant reply and the tool calls of the previous and current turn), so a follow-up like "Yes, go ahead." keeps the tools of the turn it answers.	`None`
`always_include_fallback`	`bool \| None`	Include a `request_more_tools` fallback tool when semantic selection is active. It returns the matches for a `query`, the named `tool_names`, or, called without arguments, the next `semantic_top_k` tools most relevant to the conversation; never more than :data:`MAX_TOOLS_PER_REQUEST` per call.	`None`
`embedding_model`	`str \| None`	Model name or local path for `sentence-transformers`. Defaults to `"all-MiniLM-L6-v2"` (downloaded once, then cached in `~/.cache/huggingface/`). Point to a local directory for fully-offline / air-gapped deployments:: `embedding_model="/models/all-MiniLM-L6-v2"`	`None`
`preserve_tools`	`set[str] \| None`	Tool names that are never optimized (full description and parameter descriptions are kept) and always offered by semantic selection.	`None`

ToolIndex¶

`promptise.tool_optimization.ToolIndex` ¶

In-memory semantic index over tool descriptions.

Embeds every tool's "name: description" once with sentence-transformers and ranks tools against a query by cosine similarity. build_agent(optimize_tools="semantic") builds one for you; build your own to check offline which tools a query would get before paying for any model call.

Requires pip install "promptise[tool-optimization]".

Parameters:

Name	Type	Description	Default
`tools`	`Sequence[BaseTool]`	The `BaseTool` instances to index. Duplicate names keep the first tool.	required
`model_name_or_path`	`str`	A HuggingFace model name (e.g. `"all-MiniLM-L6-v2"`) or a local directory path containing the model files for fully-offline deployments. Defaults to :data:`DEFAULT_EMBEDDING_MODEL`.	`DEFAULT_EMBEDDING_MODEL`

index = ToolIndex(agent.tools, model_name_or_path="/models/all-MiniLM-L6-v2")
names = [t.name for t in index.select("Suspend user u_42", top_k=8)]

`all_tool_names` `property` ¶

All indexed tool names.

`all_tools` `property` ¶

All indexed tools.

`tool_summaries` `property` ¶

One-line summaries of all tools (as listed by request_more_tools).

`rank(query, *, exclude=frozenset())` ¶

All indexed tool names, most relevant to query first, minus exclude.

`select(query, top_k=8, preserve=None)` ¶

Return the top_k tools most relevant to query, plus preserve.

Preserved tools are always included and don't take one of the top_k relevance slots, so top_k=3 with two preserved tools returns up to five tools. Names in preserve that aren't indexed are ignored.

Returns:

Type	Description
`list[BaseTool]`	The ranked tools (most relevant first), then the preserved ones.

`summaries(names, *, max_length=None)` ¶

- name: description lines for the indexed tools in names.

build_selection_query¶

`promptise.tool_optimization.build_selection_query(messages, *, user_turns=3)` ¶

Build the semantic-selection query from the recent conversation.

Embedding models truncate long input, so the most telling parts come first: the latest user message, the assistant reply it answers, the tool calls of the previous and current turn, then up to user_turns - 1 earlier user messages (newest first). A follow-up such as "Yes, go ahead." therefore still selects the tools of the request it confirms.

Parameters:

Name	Type	Description	Default
`messages`	`Sequence[Any]`	The conversation, as LangChain messages or `{"role": ..., "content": ...}` dicts.	required
`user_turns`	`int`	How many of the most recent user messages to include.	`3`

Returns:

Type	Description
`str`	The query text (empty if the conversation has no user message).

Adaptive Strategy¶

Failure classification and strategy learning. Agents track past failures, categorize them, and adjust behavior.

FailureCategory¶

`promptise.strategy.FailureCategory` ¶

Bases: str, Enum

Classification of a tool failure.

FailureLog¶

`promptise.strategy.FailureLog` `dataclass` ¶

A single tool failure record.

Attributes:

Name	Type	Description
`tool_name`	`str`	Name of the tool that failed.
`error_type`	`str`	Exception class name (e.g. `"ValidationError"`), or the MCP error code (e.g. `"TOOL_ERROR"`) for an MCP error result.
`error_message`	`str`	Error message (truncated to 500 chars).
`category`	`FailureCategory`	Classified failure category.
`args_preview`	`str`	Truncated preview of the failed arguments.
`timestamp`	`float`	When the failure occurred.
`confidence`	`float`	Classification confidence (0.0–1.0).
`invocation_id`	`str \| None`	Optional invocation identifier.

AdaptiveStrategyConfig¶

`promptise.strategy.AdaptiveStrategyConfig` `dataclass` ¶

Configuration for the adaptive strategy system.

Attributes:

Name	Type	Description
`enabled`	`bool`	Enable adaptive learning (default: disabled).
`synthesis_threshold`	`int`	Number of strategy failures in a scope before triggering LLM synthesis. Infrastructure failures don't count. The count is kept in memory, so it survives restarts.
`synthesis_model`	`str \| Any \| None`	LLM model ID (or chat model instance) for synthesis and verification. Defaults to the agent's own model.
`max_strategies`	`int`	Maximum stored lessons per scope. Machine lessons are dropped oldest first; human corrections only when no machine lessons are left.
`auto_cleanup`	`bool`	Delete raw failure logs after synthesis.
`strategy_ttl`	`int`	AdaptiveLesson expiry in seconds (0 = never expire).
`failure_retention`	`int`	Maximum raw failure logs kept per scope (oldest dropped first).
`verify_human_feedback`	`bool`	Use LLM-as-judge to verify human corrections that come with evidence.
`feedback_rate_limit`	`int`	Max corrections per hour per sender.
`scope`	`StrategyScope`	Who shares failures and lessons, derived from the invocation's :class:`~promptise.agent.CallerContext`: `"per_user"` (default): one partition per `CallerContext.isolation_key` (`tenant::user`). Callers without a `user_id` share one anonymous partition, which identified users never see. `"per_tenant"`: one partition per `tenant_id`; callers without a tenant fall back to `per_user`. `"per_session"`: one partition per conversation session (`chat(session_id=...)`, or `CallerContext.metadata ["session_id"]` for `ainvoke`). Without a session id nothing is recorded or injected. `"shared"`: every caller contributes to and receives the same lessons.
`confidence_half_life`	`float`	Seconds after which a synthesized lesson's confidence halves (0 = no decay). Human corrections don't decay.
`min_confidence`	`float`	Lessons whose (decayed) confidence falls below this are no longer injected and are deleted at the next synthesis.
`allowed_tools`	`Collection[str] \| None`	When set, only failures of these tools are recorded and only lessons about them are kept.
`review_lessons`	`bool`	Store synthesized lessons as pending: they are not injected until approved with :meth:`AdaptiveStrategyManager.approve_lesson`.
`learn_from_approval_denials`	`bool`	When the agent also has an approval policy, store a reviewer's denial reason as a human correction.

AdaptiveStrategyManager¶

`promptise.strategy.AdaptiveStrategyManager` ¶

Manages adaptive strategy learning for an agent.

Sits on top of a :class:MemoryProvider and handles failure recording, classification, lesson synthesis, human feedback verification, and lesson retrieval — each within the partition that :attr:AdaptiveStrategyConfig.scope derives from the current :class:~promptise.agent.CallerContext. Inside an agent invocation the caller is ambient; outside one (scripts, admin tools) pass caller= (and session_id= for per_session) explicitly.

Providers with list_entries (all built-in ones) get exact failure counts that survive restarts, failure_retention and max_strategies. A custom provider without it falls back to an in-process failure counter and best-effort, search-based limits.

Parameters:

Name	Type	Description	Default
`config`	`AdaptiveStrategyConfig`	Adaptive strategy configuration.	required
`memory`	`Any`	The agent's memory provider for storing strategies.	required
`agent_model`	`str \| Any \| None`	Default LLM model ID for synthesis/verification.	`None`
`guardrails`	`Any \| None`	Optional guardrails scanner for feedback validation.	`None`
`tool_names`	`Iterable[str] \| None`	Names of the agent's tools. A synthesized lesson that names another tool than the one it is about is rejected.	`None`

`config` `property` ¶

The adaptive strategy configuration.

`tool_names` `property` ¶

The agent's tool names, used to vet synthesized lessons.

`approve_lesson(lesson_id, *, caller=_UNSET, session_id=None)` `async` ¶

Activate a pending lesson. Returns its new id, or None if not found.

Memory providers have no update, so the lesson is re-stored as active and the pending entry deleted.

`drain()` `async` ¶

Wait for background work (approval-denial learning) to finish.

`forget_lesson(lesson_id, *, caller=_UNSET, session_id=None)` `async` ¶

Delete one of the current scope's lessons (active or pending).

`format_strategy_block(strategies)` ¶

Format strategies for injection into the agent's system prompt.

Wraps in <strategy_context> fences with anti-injection disclaimer.

`get_relevant_strategies(query, *, limit=3, caller=_UNSET, session_id=None)` `async` ¶

Find the current scope's lessons relevant to query.

Lessons are found by the provider's search, topped up (for providers with list_entries) with the scope's lessons that share words with the query. Returns highest (decayed) confidence first; expired, pending and low-confidence lessons are excluded.

`list_lessons(*, caller=_UNSET, session_id=None, include_pending=True)` `async` ¶

The current scope's lessons, oldest first.

`pending_lessons(*, caller=_UNSET, session_id=None)` `async` ¶

Synthesized lessons awaiting review (review_lessons=True).

`record_approval_denial(request, decision)` `async` ¶

Store a reviewer's denial reason as a human correction.

Called for approval decisions with approved=False and a reason; denials without a reason teach nothing and are skipped. Timeouts and handler errors never reach here.

`record_failure(failure, *, caller=_UNSET, session_id=None)` `async` ¶

Record a tool failure in the current caller's scope.

Infrastructure failures (MCP server down, network errors) are skipped — the agent shouldn't learn from infra problems. Unknown failures are stored with low confidence and don't count toward synthesis. Reaching synthesis_threshold strategy failures in the scope triggers :meth:synthesize.

Parameters:

Name	Type	Description	Default
`failure`	`FailureLog`	The failure to record.	required
`caller`	`CallerContext \| None`	Whose failure this is. Defaults to the ambient :class:`~promptise.agent.CallerContext` of the invocation.	`_UNSET`
`session_id`	`str \| None`	Session for `scope="per_session"`. Defaults to the current `chat()` session.	`None`

`record_human_correction(correction, *, evidence=None, sender_id=None, tool_name=None, caller=_UNSET, session_id=None, source='human_feedback')` `async` ¶

Process a human correction ("you did this wrong").

Validates the correction against guardrails and optionally verifies it via LLM-as-judge before storing it in the current scope. Human corrections rank above synthesized lessons: confidence 0.9 unverified, 1.0 when the judge confirms it, 0.4 when the judge rejects it.

Parameters:

Name	Type	Description	Default
`correction`	`str`	The human's feedback text.	required
`evidence`	`dict[str, Any] \| None`	Tool call history and output for verification.	`None`
`sender_id`	`str \| None`	Who sent the correction (for rate limiting + audit).	`None`
`tool_name`	`str \| None`	The tool the correction is about, if any.	`None`
`caller`	`CallerContext \| None`	Whose scope to store it in (default: ambient caller).	`_UNSET`
`session_id`	`str \| None`	Session for `scope="per_session"`.	`None`
`source`	`str`	Recorded as the lesson's source.	`'human_feedback'`

Returns:

Type	Description
`bool`	`True` if the correction was accepted, `False` if rejected.

`reset(*, caller=_UNSET, session_id=None)` `async` ¶

Delete every lesson, failure log and counter in the current scope.

`synthesize(*, caller=_UNSET, session_id=None)` `async` ¶

Synthesize lessons from the scope's unsynthesized failure logs now.

Asks the LLM to reflect on recent failures and propose one-sentence lessons, each about how to call one of the failing tools. Failure text is passed as untrusted data, and every proposed lesson is vetted: it must be about a failing tool, must not name another tool, and must not contain a URL, email address or IP address. Rejected lessons are logged. Returns the number stored.

AdaptiveLesson¶

`promptise.strategy.AdaptiveLesson` `dataclass` ¶

A stored lesson, as returned by :meth:AdaptiveStrategyManager.list_lessons.

Attributes:

Name	Type	Description
`id`	`str`	Memory id (pass to `approve_lesson` / `forget_lesson`).
`text`	`str`	The lesson as injected.
`tool`	`str \| None`	The tool it is about, when known.
`source`	`str`	`"synthesis"`, `"human_feedback"` or `"approval_denial"`.
`confidence`	`float`	Current confidence, after decay.
`status`	`str`	`"active"` (injected) or `"pending"` (awaiting review).
`created`	`float`	Unix timestamp.

classify_failure¶

`promptise.strategy.classify_failure(error_type, error_message)` ¶

Classify an error as infrastructure, strategy, or unknown.

Deterministic — no LLM call. error_type is an exception class name or an MCP error code. Rules, first match wins:

Infrastructure exception types (ConnectionError, TimeoutError, MCPClientError…) and codes (RATE_LIMIT_EXCEEDED, TIMEOUT, INTERNAL_ERROR, AUTHENTICATION_ERROR…).
Infrastructure phrases in the message, as whole words: an HTTP 5xx/429 status next to an HTTP word or reason phrase, rate limits, refused/reset connections, timeouts, invalid credentials.
Strategy exception types (ValueError, ValidationError…) and codes (TOOL_ERROR, VALIDATION_ERROR, ACCESS_DENIED…) — a tool that deliberately rejected the call.
Strategy phrases ("not found", "invalid", "already booked"…).
Otherwise :attr:FailureCategory.UNKNOWN.

Parameters:

Name	Type	Description	Default
`error_type`	`str`	The exception class name or MCP error code.	required
`error_message`	`str`	The error message text.	required

Returns:

Type	Description
`FailureCategory`	The classified :class:`FailureCategory`.

Context Engine¶

Token budgeting across multiple context layers with model-aware window detection.

ContextLayer¶

`promptise.context_engine.ContextLayer` `dataclass` ¶

A single context source in the assembly pipeline.

Attributes:

Name	Type	Description
`name`	`str`	Unique layer identifier (e.g. `"memory"`, `"strategies"`).
`priority`	`int`	Assembly priority (0-10). Higher = kept longer when trimming. Required layers (priority >= 9) are never dropped.
`content`	`str`	The text content for this layer.
`required`	`bool`	If `True`, this layer is never trimmed (overrides priority).
`trim_strategy`	`str`	How to trim this layer — `"truncate"` (cut from end) or `"oldest_first"` (for conversation history, removes oldest messages first).
`metadata`	`dict[str, Any]`	Developer-provided metadata for reporting.

`token_estimate` `property` ¶

Rough token estimate (chars / 3.5) for quick checks.

ContextReport¶

`promptise.context_engine.ContextReport` `dataclass` ¶

Report of context assembly — what was included, trimmed, and total usage.

Attributes:

Name	Type	Description
`total_tokens`	`int`	Total tokens in the assembled context.
`budget`	`int`	The model's context window minus response reserve.
`layers`	`list[dict[str, Any]]`	Per-layer token counts and trim status.
`trimmed_layers`	`list[str]`	Layers that were trimmed or dropped.
`utilization`	`float`	Fraction of budget used (0.0-1.0).

`utilization` `property` ¶

Fraction of budget used.

Tokenizer¶

`promptise.context_engine.Tokenizer` ¶

Bases: Protocol

Protocol for token counters.

Implementations must provide a count(text) -> int method that returns the exact (or best-estimate) token count for a string.

Callback Handler¶

Bridges LangChain callbacks into the observability collector.

PromptiseCallbackHandler¶

`promptise.callback_handler.PromptiseCallbackHandler` ¶

Bases: BaseCallbackHandler

LangChain callback handler → ObservabilityCollector bridge.

Captures every LLM turn, tool call, token count, latency, retry, and error. Designed to be the only integration point needed between LangChain's event system and Promptise's observability.

What is recorded depends on level (see :class:~promptise.observability_config.ObserveLevel): OFF records nothing, BASIC skips the per-turn llm.start / llm.end events (token usage is still added to the current agent run), STANDARD records every turn, and FULL also records prompt and response text unless record_prompts=False.

Parameters:

Name	Type	Description	Default
`collector`	`Any`	The :class:`~promptise.observability.ObservabilityCollector`.	required
`agent_id`	`str \| None`	Agent identifier stamped on every event.	`None`
`record_prompts`	`bool \| None`	Record prompt and response text. `None` follows `level` (on at `FULL` only); `True` / `False` force it.	`None`
`level`	`ObserveLevel`	Detail level.	`STANDARD`
`record_tool_io`	`bool`	Record tool arguments and result previews. When `False`, only their lengths are recorded.	`True`

`get_summary()` ¶

Return a summary of all tracked metrics.

`on_llm_new_token(token, *, run_id, **kwargs)` ¶

Capture streaming tokens (FULL level only).

token is typed to match LangChain-core's LLMManagerMixin base, which widened it from str to str | list[str | dict] in the 1.x line. The wider annotation satisfies the override contract on both the 0.3.x and 1.x major versions.

Exceptions¶

SuperAgentError¶

`promptise.exceptions.SuperAgentError` ¶

Bases: RuntimeError

Base exception for all SuperAgent-related errors.

This is the base class for all exceptions raised by the .superagent file loading and validation system. Catching this exception will catch all SuperAgent-specific errors.

Examples:

>>> raise SuperAgentError("Failed to load configuration")

SuperAgentValidationError¶

`promptise.exceptions.SuperAgentValidationError` ¶

Bases: SuperAgentError

Raised when .superagent schema validation fails.

This exception is raised when a .superagent file fails Pydantic schema validation. It includes detailed error information to help users fix the configuration.

Attributes:

Name	Type	Description
`errors`		List of Pydantic validation errors with location and message.
`file_path`		Path to the file that failed validation.

Examples:

>>> raise SuperAgentValidationError(
...     "Schema validation failed",
...     errors=[{"loc": ["agent", "model"], "msg": "field required"}],
...     file_path="/path/to/agent.superagent"
... )

`init(message, *, errors=None, file_path=None)` ¶

Initialize validation error with details.

Parameters:

Name	Type	Description	Default
`message`	`str`	Human-readable error message.	required
`errors`	`list[dict[str, Any]] \| None`	List of Pydantic validation errors (from ValidationError.errors()).	`None`
`file_path`	`str \| None`	Path to the configuration file that failed validation.	`None`

`str()` ¶

Format validation errors for display.

Returns:

Type	Description
`str`	Formatted error message including file path and detailed validation errors.

EnvVarNotFoundError¶

`promptise.exceptions.EnvVarNotFoundError` ¶

Bases: SuperAgentError

Raised when a required environment variable is not found.

This exception is raised when a configuration references an environment variable using ${VAR_NAME} syntax, but the variable is not set in the environment.

Attributes:

Name	Type	Description
`var_name`		The name of the missing environment variable.
`context`		Optional context about where the variable was referenced (e.g., "servers.math.url").

Examples:

>>> raise EnvVarNotFoundError("OPENAI_API_KEY", context="agent.model.api_key")

`init(var_name, context=None)` ¶

Initialize environment variable error.

Parameters:

Name	Type	Description	Default
`var_name`	`str`	Name of the missing environment variable.	required
`context`	`str \| None`	Optional context string indicating where the variable was referenced (e.g., field path in configuration).	`None`

Type Definitions¶

ModelLike¶

`promptise.types.ModelLike = str | Model | BaseChatModel | Runnable[Any, Any]` `module-attribute` ¶

System Prompt¶

DEFAULT_SYSTEM_PROMPT¶

`promptise.prompt.DEFAULT_SYSTEM_PROMPT = 'You are a capable deep agent. Use available tools from connected MCP servers to plan and execute tasks. Always inspect tool descriptions and input schemas before calling them. Be precise and avoid hallucinating tool arguments. Prefer calling tools rather than guessing, and cite results from tools clearly.'` `module-attribute` ¶

Utilities API Reference¶

Environment Variables¶

resolve_env_var¶

promptise.env_resolver.resolve_env_var(value, *, context=None, allow_missing=False) ¶

resolve_env_in_dict¶

promptise.env_resolver.resolve_env_in_dict(data, *, context_prefix='') ¶

validate_all_env_vars_available¶

promptise.env_resolver.validate_all_env_vars_available(data) ¶

MCP Tool Discovery¶

ToolInfo¶

promptise.tools.ToolInfo dataclass ¶

MCPClientError¶

promptise.tools.MCPClientError ¶

Approval (Human-in-the-loop)¶

ApprovalRequest¶

promptise.approval.ApprovalRequest dataclass ¶

compute_hmac(secret) ¶

to_dict() ¶

ApprovalDecision¶

promptise.approval.ApprovalDecision dataclass ¶

ApprovalHandler¶

promptise.approval.ApprovalHandler ¶

request_approval(request) async ¶

CallbackApprovalHandler¶

promptise.approval.CallbackApprovalHandler ¶

request_approval(request) async ¶

WebhookApprovalHandler¶

promptise.approval.WebhookApprovalHandler ¶

request_approval(request) async ¶

QueueApprovalHandler¶

promptise.approval.QueueApprovalHandler ¶

request_approval(request) async ¶

submit_decision(request_id, decision) ¶

ApprovalPolicy¶

promptise.approval.ApprovalPolicy ¶

record_decision(request, decision) async ¶

redact_arguments(arguments) async ¶

request_metadata(tool_name, arguments) async ¶

requires_approval(tool_name) ¶

summarize_context(messages) async ¶

approval_elicitation_callback¶

promptise.approval.approval_elicitation_callback(handler, *, server_name=None, in_flight=None, timeout=None, event_notifier=None) ¶

Semantic Cache¶

SemanticCache¶

promptise.cache.SemanticCache ¶

allows_conversation(messages) ¶

check(query_text, *, context_fingerprint='', caller=None, model_id=None, instruction_hash='') async ¶

check_dependencies() ¶

close() async ¶

invalidate_for_write(tool_name, caller=None) async ¶

is_write_tool(name, annotations=None) ¶

purge_user(user_id, *, tenant_id=None) async ¶

stats() async ¶

store(query_text, response_text, output, *, context_fingerprint='', caller=None, model_id=None, instruction_hash='', tools_used=None, write_generation=None) async ¶

warmup() ¶

write_generation(caller=None) ¶

CacheEntry¶

promptise.cache.CacheEntry dataclass ¶

age property ¶

expired property ¶

verify_checksum() ¶

CacheStats¶

promptise.cache.CacheStats dataclass ¶

EmbeddingProvider¶

promptise.cache.EmbeddingProvider ¶

LocalEmbeddingProvider¶

promptise.cache.LocalEmbeddingProvider ¶

embed(texts) async ¶

warmup() ¶

OpenAIEmbeddingProvider¶

promptise.cache.OpenAIEmbeddingProvider ¶

embed(texts) async ¶

InMemoryCacheBackend¶

promptise.cache.InMemoryCacheBackend ¶

invalidate(scope_key, pattern=None) async ¶

purge_user(user_id) async ¶

search(scope_key, embedding, threshold, *, match=None) async ¶

store(scope_key, entry) async ¶

RedisCacheBackend¶

promptise.cache.RedisCacheBackend ¶

`promptise.env_resolver.resolve_env_var(value, *, context=None, allow_missing=False)` ¶

`promptise.env_resolver.resolve_env_in_dict(data, *, context_prefix='')` ¶

`promptise.env_resolver.validate_all_env_vars_available(data)` ¶

`promptise.tools.ToolInfo` `dataclass` ¶

`promptise.tools.MCPClientError` ¶

`promptise.approval.ApprovalRequest` `dataclass` ¶

`compute_hmac(secret)` ¶

`to_dict()` ¶

`promptise.approval.ApprovalDecision` `dataclass` ¶

`promptise.approval.ApprovalHandler` ¶

`request_approval(request)` `async` ¶

`promptise.approval.CallbackApprovalHandler` ¶

`request_approval(request)` `async` ¶

`promptise.approval.WebhookApprovalHandler` ¶

`request_approval(request)` `async` ¶

`promptise.approval.QueueApprovalHandler` ¶

`request_approval(request)` `async` ¶

`submit_decision(request_id, decision)` ¶

`promptise.approval.ApprovalPolicy` ¶

`record_decision(request, decision)` `async` ¶

`redact_arguments(arguments)` `async` ¶

`request_metadata(tool_name, arguments)` `async` ¶

`requires_approval(tool_name)` ¶

`summarize_context(messages)` `async` ¶

`promptise.approval.approval_elicitation_callback(handler, *, server_name=None, in_flight=None, timeout=None, event_notifier=None)` ¶

`promptise.cache.SemanticCache` ¶

`allows_conversation(messages)` ¶

`check(query_text, *, context_fingerprint='', caller=None, model_id=None, instruction_hash='')` `async` ¶

`check_dependencies()` ¶

`close()` `async` ¶

`invalidate_for_write(tool_name, caller=None)` `async` ¶

`is_write_tool(name, annotations=None)` ¶

`purge_user(user_id, *, tenant_id=None)` `async` ¶

`stats()` `async` ¶

`store(query_text, response_text, output, *, context_fingerprint='', caller=None, model_id=None, instruction_hash='', tools_used=None, write_generation=None)` `async` ¶

`warmup()` ¶

`write_generation(caller=None)` ¶

`promptise.cache.CacheEntry` `dataclass` ¶

`age` `property` ¶

`expired` `property` ¶

`verify_checksum()` ¶

`promptise.cache.CacheStats` `dataclass` ¶

`promptise.cache.EmbeddingProvider` ¶

`promptise.cache.LocalEmbeddingProvider` ¶

`embed(texts)` `async` ¶

`warmup()` ¶

`promptise.cache.OpenAIEmbeddingProvider` ¶

`embed(texts)` `async` ¶

`promptise.cache.InMemoryCacheBackend` ¶

`invalidate(scope_key, pattern=None)` `async` ¶

`purge_user(user_id)` `async` ¶

`search(scope_key, embedding, threshold, *, match=None)` `async` ¶

`store(scope_key, entry)` `async` ¶

`promptise.cache.RedisCacheBackend` ¶

`close()` `async` ¶

`invalidate(scope_key, pattern=None)` `async` ¶

`purge_user(user_id)` `async` ¶

`search(scope_key, embedding, threshold, *, match=None)` `async` ¶

`store(scope_key, entry)` `async` ¶

`promptise.events.AgentEvent` `dataclass` ¶

`compute_hmac(secret)` ¶

`to_dict()` ¶

`promptise.events.EventSink` ¶

`emit(event)` `async` ¶

`promptise.events.EventNotifier` ¶

`is_running` `property` ¶

`sinks` `property` ¶

`add_sink(sink)` ¶

`emit(event)` `async` ¶

`emit_sync(event)` ¶

`flush(timeout=None)` `async` ¶

`start()` `async` ¶

`stop(timeout=None)` `async` ¶

`promptise.events.WebhookSink` ¶

`secret` `property` ¶

`close()` `async` ¶

`emit(event)` `async` ¶

`promptise.events.CallbackSink` ¶

`emit(event)` `async` ¶

`promptise.events.LogSink` ¶