Interface API

Public developer‑facing modules under actingweb.interface.

ActingWebApp

Main ActingWebApp class providing fluent API for application configuration.

class actingweb.interface.app.ActingWebApp(aw_type: str, database: str | None = None, fqdn: str = '', proto: str = 'https://')[source]

Bases: object

Main application class for ActingWeb with fluent configuration API.

Example usage:

app = (
    ActingWebApp(
        aw_type="urn:actingweb:example.com:myapp",
        database="dynamodb",
        fqdn="myapp.example.com",
    )
    .with_oauth(client_id="...", client_secret="...")
    .with_web_ui()
    .with_devtest()
)

@app.lifecycle_hook("actor_created")
def handle_actor_created(actor: 'ActorInterface') -> None:
    # Custom logic after actor creation
    pass
action_hook(action_name: str = '*', description: str = '', input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None) → Callable[[...], Any][source]

Decorator to register action hooks with optional metadata.

Parameters:

  • action_name – Name of action to hook (“*” for all actions)

  • description – Human-readable description of what the action does

  • input_schema – JSON schema describing expected input parameters

  • output_schema – JSON schema describing the expected return value. This describes the ActingWeb /actions / /methods discovery surface only. It is NOT forwarded to MCP: a tool’s outputSchema comes solely from @mcp_tool(output_schema=...). And neither one causes structuredContent to be emitted – only a structuredContent key in the returned dict does that.

  • annotations – Safety/behavior hints (e.g., destructiveHint, readOnlyHint)

add_actor_type(name: str, factory: str = '', relationship: str = 'friend') → ActingWebApp[source]

Add an actor type configuration.

add_box(client_id: str, client_secret: str) → ActingWebApp[source]

Add Box service using pre-configured template.

add_dropbox(client_id: str, client_secret: str) → ActingWebApp[source]

Add Dropbox service using pre-configured template.

add_github(client_id: str, client_secret: str) → ActingWebApp[source]

Add GitHub service using pre-configured template.

add_gmail(client_id: str, client_secret: str, readonly: bool = True) → ActingWebApp[source]

Add Gmail service using pre-configured template.

add_service(name: str, client_id: str, client_secret: str, scopes: list, auth_uri: str, token_uri: str, userinfo_uri: str = '', revocation_uri: str = '', base_api_url: str = '', **extra_params) → ActingWebApp[source]

Add a custom third-party OAuth2 service configuration.

app_callback_hook(callback_name: str) → Callable[[...], Any][source]

Decorator to register application-level callback hooks (no actor context).

callback_hook(callback_name: str = '*') → Callable[[...], Any][source]

Decorator to register actor-level callback hooks.

get_config() → Config[source]

Get the underlying ActingWeb Config object.

get_service_registry()[source]

Get the service registry for advanced configuration.

get_subscription_config() → SubscriptionProcessingConfig[source]

Get the subscription processing configuration.

integrate_fastapi(fastapi_app: Any, templates_dir: str | None = None, **options: Any) → FastAPIIntegration[source]

Integrate ActingWeb with FastAPI application.

Parameters:

  • fastapi_app – The FastAPI application instance

  • templates_dir – Directory containing Jinja2 templates (optional)

  • **options – Additional configuration options

Returns:

FastAPIIntegration instance

Raises:

ImportError – If FastAPI is not installed

integrate_flask(flask_app: Any) → FlaskIntegration[source]

Integrate with Flask application.

is_mcp_enabled() → bool[source]

Check if MCP functionality is enabled.

lifecycle_hook(event: str) → Callable[[...], Any][source]

Decorator to register lifecycle hooks.

method_hook(method_name: str = '*', description: str = '', input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None) → Callable[[...], Any][source]

Decorator to register method hooks with optional metadata.

Parameters:

  • method_name – Name of method to hook (“*” for all methods)

  • description – Human-readable description of what the method does

  • input_schema – JSON schema describing expected input parameters

  • output_schema – JSON schema describing the expected return value. This describes the ActingWeb /actions / /methods discovery surface only. It is NOT forwarded to MCP: a tool’s outputSchema comes solely from @mcp_tool(output_schema=...). And neither one causes structuredContent to be emitted – only a structuredContent key in the returned dict does that.

  • annotations – Safety/behavior hints (e.g., readOnlyHint, idempotentHint)

property_hook(property_name: str = '*') → Callable[[...], Any][source]

Decorator to register property hooks.

run(host: str = '0.0.0.0', port: int = 5000, debug: bool = False) → None[source]

Run as standalone application with Flask.

subscription_data_hook(target: str = '*') → Callable[[...], Any][source]

Decorator to register subscription data hooks.

Use with .with_subscription_processing() for automatic handling. The handler receives already-sequenced, deduplicated data.

Parameters:

target – Target to hook (e.g., “properties”, “*” for all)

Example:

@app.subscription_data_hook("properties")
def on_property_change(
    actor: ActorInterface,
    peer_id: str,
    target: str,
    data: dict,
    sequence: int,
    callback_type: str
) -> None:
    # Data is already sequenced and stored
    pass
subscription_hook(func: Callable[[...], Any]) → Callable[[...], Any][source]

Decorator to register subscription hooks.

with_apple_sign_in(client_id: str, *, audiences: list[str], team_id: str, key_id: str, private_key_path: str | None = None, private_key_pem: str | None = None, scope: str = 'openid name email', web_redirect_uri: str = '', mobile_redirect_uri: str = '') → ActingWebApp[source]

Configure Sign in with Apple as an OAuth login provider.

Apple differs from Google/GitHub: the client_secret is a freshly signed ES256 JWT (Team ID + Key ID + .p8), there is no userinfo endpoint (identity comes from the id_token), and the callback is a POST (response_mode=form_post).

Parameters:

  • client_id – The Apple Services ID (web/Android) used as client_id.

  • audiences – Acceptable aud values for id_token validation — typically the Services ID plus the iOS Bundle ID. Must be non-empty.

  • team_id – Apple Developer Team ID (the JWT iss).

  • key_id – Key ID of the Sign in with Apple .p8 key.

  • private_key_path – Path to the .p8 file. Takes precedence over private_key_pem and over the APPLE_PRIVATE_KEY_PATH env var.

  • private_key_pem – PEM string of the .p8 key (falls back to the APPLE_PRIVATE_KEY_PEM env var).

  • scope – OAuth scope. Defaults to "openid name email".

  • web_redirect_uri – Override for the web/SPA Apple redirect_uri (must be HTTPS; defaults to /oauth/callback/apple).

  • mobile_redirect_uri – If set, also registers an apple-mobile provider whose deep-link redirect the Capacitor app intercepts.

Raises:

ValueError – If required fields are missing or the .p8 key cannot be loaded/parsed (validated eagerly here, not at first request).

with_bot(token: str = '', email: str = '', secret: str = '', admin_room: str = '') → ActingWebApp[source]

Configure bot integration.

with_devtest(enable: bool = True) → ActingWebApp[source]

Enable or disable development/testing endpoints.

with_dynamodb(auto_create_tables: bool = True) → ActingWebApp[source]

Configure DynamoDB backend options.

Parameters:

auto_create_tables – When True (default), tables are created automatically on first access — convenient for development. Set False for production deployments that manage tables via infrastructure-as-code; the runtime IAM role can then drop the dynamodb:CreateTable and dynamodb:DescribeTable permissions.

Returns:

Self for method chaining

Note

An explicit call takes precedence over the AWS_DB_AUTO_CREATE_TABLES environment variable. Call it before integrating with the web framework — the setting only affects tables not yet accessed. Has no effect on the PostgreSQL backend (tables there are managed by Alembic).

with_email_as_creator(enable: bool = True) → ActingWebApp[source]

Force email property as creator.

with_github(client_id: str, client_secret: str, *, scope: str = 'read:user user:email', redirect_uri: str = '', mobile_redirect_uri: str = '') → ActingWebApp[source]

Configure GitHub sign-in, with optional native-mobile support.

Ergonomic equivalent of with_oauth(provider="github", ...) that fills in GitHub’s authorization/token endpoints. When mobile_redirect_uri is set it also registers a github-mobile provider whose OAuth redirect_uri stays on the HTTPS /oauth/callback so the authorization code is exchanged server-side and the Capacitor app only ever receives an opaque single-use ticket on mobile_redirect_uri (the mobile_ticket grant). The code therefore never rides a hijackable custom-scheme redirect.

GitHub issues no OIDC id_token; identity comes from the /user API after the code exchange, so GitHub mobile uses this ticket flow rather than the JWT-bearer grant used by Apple/Google native.

Parameters:

  • client_id – GitHub OAuth app client ID.

  • client_secret – GitHub OAuth app client secret.

  • scope – OAuth scope. Defaults to "read:user user:email".

  • redirect_uri – Override for the web redirect_uri (defaults to /oauth/callback).

  • mobile_redirect_uri – Custom-scheme deep link the Capacitor app intercepts; when set, registers the github-mobile provider.

with_google_native(client_id: str, *, audiences: list[str] | None = None, web_client_id: str | None = None, ios_client_id: str | None = None, android_client_id: str | None = None, android_server_client_id: str | None = None, client_secret: str = '', scope: str = 'openid profile email', redirect_uri: str = '') → ActingWebApp[source]

Configure Google native sign-in via the JWT-bearer grant.

Native iOS/Android Google sign-in yields an id_token the app submits to POST /oauth/spa/token with the jwt-bearer grant. The library validates it against Google’s JWKS, accepting any of the configured audiences (the per-platform client IDs).

Parameters:

  • client_id – Primary client ID (also added to audiences).

  • audiences – Explicit acceptable aud values. If omitted, built from client_id plus the four *_client_id kwargs.

  • / (web_client_id / ios_client_id / android_client_id) – android_server_client_id: Optional per-platform client IDs folded into audiences.

  • client_secret – Optional; only needed if this provider also performs a code exchange (not required for the pure JWT-bearer path).

  • scope – OAuth scope.

  • redirect_uri – Optional redirect override.

Raises:

ValueError – If no audiences can be derived.

with_indexed_properties(properties: list[str] | None = None) → ActingWebApp[source]

Configure which properties support reverse lookups via lookup table.

Properties specified here will have their values indexed in a separate lookup table, enabling reverse lookups (value -> actor_id) without the 2048-byte size limit imposed by DynamoDB Global Secondary Indexes.

Parameters:

properties – List of property names to index. Default is [“oauthId”, “email”, “externalUserId”]. Set to empty list [] to disable all reverse lookups.

Returns:

Self for method chaining

Example:

app = (
    ActingWebApp(...)
    .with_indexed_properties(["oauthId", "email", "customUserId"])
)

Note

Only properties listed here can be used with Actor.get_from_property(). Changes require application restart to take effect. An explicit call to this method takes precedence over the INDEXED_PROPERTIES environment variable; if this method is never called, the env variable (when set) overrides the library default.

with_legacy_property_index(enable: bool = False) → ActingWebApp[source]

Enable legacy GSI/index-based property reverse lookup (for migration).

When False, uses the new lookup table approach which supports property values larger than 2048 bytes. When True, uses the legacy DynamoDB GSI or PostgreSQL index on the value field (limited to 2048 bytes).

Parameters:

enable – True to use legacy GSI/index, False for new lookup table

Returns:

Self for method chaining

Note

An explicit call to this method takes precedence over the USE_PROPERTY_LOOKUP_TABLE environment variable. If this method is never called, the env variable (when set) overrides the library default, which is currently the legacy GSI/index path.

with_mcp(enable: bool = True, server_name: str = 'actingweb', instructions: str | None = None) → ActingWebApp[source]

Enable or disable MCP (Model Context Protocol) functionality.

Parameters:

  • enable – If True, enable MCP support.

  • server_name – Name announced in the MCP initialise handshake and reported by GET /mcp and GET /mcp/info. Some clients use this as the default tool prefix (emm:search vs actingweb:search). Defaults to "actingweb". The first with_mcp() call sets the process-wide singleton name; subsequent re-configuration does not rename existing per-actor servers.

  • instructions – Optional server-level orientation string surfaced on the MCP InitializeResult.instructions field per protocol. Clients display it to the LLM on initial connection. Use it to point new LLMs at an entry-point tool (e.g. how_to_use()). Like server_name, the first call wins for the singleton.

with_oauth(client_id: str, client_secret: str, scope: str = '', auth_uri: str = '', token_uri: str = '', provider: str = '', **kwargs: Any) → ActingWebApp[source]

Configure OAuth authentication.

Can be called multiple times with different provider values to configure multiple OAuth providers simultaneously.

Note

When calling with_oauth() multiple times for different providers, every call must include an explicit provider name. Mixing a nameless call (legacy single-provider API) with named calls will silently drop the nameless provider from the multi-provider configuration.

Parameters:

  • client_id – OAuth client ID

  • client_secret – OAuth client secret

  • scope – OAuth scope string

  • auth_uri – Authorization endpoint URL

  • token_uri – Token exchange endpoint URL

  • provider – Provider name (e.g. "google", "github"). When empty, stores as the single default provider for backward compatibility.

  • **kwargs – Additional OAuth config values

with_peer_capabilities(enable: bool = True, max_age_seconds: int = 3600) → ActingWebApp[source]

Enable peer capabilities (methods/actions) caching for trust relationships.

When enabled, peer methods and actions are automatically fetched and cached when trust relationships are established. Capabilities are refreshed during sync_peer() operations only if the cache is stale (older than max_age_seconds).

Parameters:

  • enable – Whether to enable capabilities caching. Default True.

  • max_age_seconds – Maximum age in seconds before cached capabilities are considered stale and refetched. Default 3600 (1 hour). Capabilities (methods/actions) rarely change, so 1 hour is conservative. Set to 0 to always refetch.

Returns:

Self for method chaining.

Example:

app = (
    ActingWebApp(...)
    .with_peer_capabilities(enable=True, max_age_seconds=7200)
)

# Access via TrustManager
capabilities = actor.trust.get_peer_capabilities(peer_id)
if capabilities:
    for method in capabilities.methods:
        print(f"Method: {method.name} - {method.description}")
with_peer_permissions(enable: bool = True, auto_delete_on_revocation: bool = False, notify_peer_on_change: bool = True) → ActingWebApp[source]

Enable peer permissions caching for trust relationships.

When enabled, peer permissions are automatically fetched and cached when trust relationships are established. Permissions are refreshed during sync_peer() operations.

This caches what permissions the REMOTE peer has granted US access to. It is distinct from TrustPermissions which stores what WE grant to peers.

Permission callbacks from peers are automatically processed and stored when this is enabled.

Parameters:

  • enable – Whether to enable permissions caching. Default True.

  • auto_delete_on_revocation – When True, automatically delete cached peer data from RemotePeerStore when the peer revokes property access. This ensures that when a peer revokes access to certain data (e.g., memory_* properties), the locally cached copies are deleted. Default False.

  • notify_peer_on_change – When True (default), automatically notify peers when their permissions change. This sends a callback to the peer’s /callbacks/permissions/{actor_id} endpoint. The notification is fire-and-forget (failures logged but don’t block the store operation).

Returns:

Self for method chaining.

Example:

app = (
    ActingWebApp(...)
    .with_peer_permissions(
        enable=True,
        auto_delete_on_revocation=True,  # Delete cached data on revocation
        notify_peer_on_change=True       # Auto-notify peers (default)
    )
)

# Access cached permissions via PeerPermissionStore
from actingweb.peer_permissions import get_peer_permission_store
store = get_peer_permission_store(actor.config)
permissions = store.get_permissions(actor.id, peer_id)
if permissions:
    if permissions.has_property_access("memory_travel", "read"):
        print("Peer granted us access to memory_travel")
with_peer_profile(attributes: list[str] | None = None) → ActingWebApp[source]

Enable peer profile caching for trust relationships.

When enabled, peer profile attributes are automatically fetched and cached when trust relationships are established. Profiles are refreshed during sync_peer() operations.

Parameters:

attributes – List of property names to cache from peer actors. Default: [“displayname”, “email”, “description”] Pass empty list to explicitly disable caching.

Returns:

Self for method chaining.

Example:

app = (
    ActingWebApp(...)
    .with_peer_profile(attributes=["displayname", "email", "avatar_url"])
)

# Access via TrustManager
profile = actor.trust.get_peer_profile(peer_id)
if profile:
    print(f"Connected with {profile.displayname}")
with_spa_cors_origins(*origins: str) → ActingWebApp[source]

Restrict the CORS origins allowed on the SPA OAuth endpoints.

Controls Access-Control-Allow-Origin for the /oauth/spa/* endpoints. The default is "*" (echo the request origin — allow all), which is convenient for development but should be tightened in production to the specific origins that serve your SPA. Calling with no arguments resets to allow-all.

Origins are scheme + host (+ optional port), e.g. https://app.example.com.

Parameters:

*origins – One or more allowed CORS origins. No arguments restores the allow-all ("*") default.

Returns:

Self for method chaining

Example:

app = (
    ActingWebApp(...)
    .with_spa_cors_origins(
        "https://app.example.com",
        "https://staging.app.example.com",
    )
)
with_spa_redirect_origins(*origins: str) → ActingWebApp[source]

Allow additional SPA redirect origins (split-domain deployments).

The redirect_uri passed to POST /oauth/spa/authorize is validated against an allowlist — the backend’s own FQDN plus the origins of configured OAuth redirect URIs / Apple mobile deep links. An off-origin redirect_uri is rejected with 400 (closing an open-redirect / one-time-session-id leak). Use this when your SPA is served from a different origin than the backend FQDN so its authorize requests are accepted.

Origins must be scheme + host (+ optional port), e.g. https://app.example.com. Same-origin SPAs need no configuration.

Parameters:

*origins – One or more allowed SPA origins. Calling with no arguments clears any previously configured origins.

Returns:

Self for method chaining

Example:

app = (
    ActingWebApp(...)
    .with_spa_redirect_origins("https://app.example.com")
)
with_subscription_processing(auto_sequence: bool = True, auto_storage: bool = True, auto_cleanup: bool = True, gap_timeout_seconds: float = 5.0, max_pending: int = 100, storage_prefix: str = 'remote:', max_concurrent_callbacks: int = 10, max_payload_for_high_granularity: int = 65536, circuit_breaker_threshold: int = 5, circuit_breaker_cooldown: float = 60.0) → ActingWebApp[source]

Enable automatic subscription processing.

When enabled, the library automatically handles: - Callback sequencing and deduplication - Gap detection and resync triggering - Data storage in RemotePeerStore (if auto_storage=True) - Cleanup when trust is deleted (if auto_cleanup=True)

Parameters:

  • auto_sequence – Enable CallbackProcessor for sequence handling

  • auto_storage – Automatically store received data in RemotePeerStore

  • auto_cleanup – Register hook to clean up when trust is deleted

  • gap_timeout_seconds – Time before triggering resync on sequence gap

  • max_pending – Maximum pending callbacks before back-pressure (429)

  • storage_prefix – Bucket prefix for RemotePeerStore

  • max_concurrent_callbacks – Max concurrent callback deliveries

  • max_payload_for_high_granularity – Payload size before granularity downgrade

  • circuit_breaker_threshold – Failures before opening circuit

  • circuit_breaker_cooldown – Seconds before testing recovery

Returns:

Self for method chaining

with_sync_callbacks(enable: bool = True) → ActingWebApp[source]

Enable synchronous subscription callbacks.

When enabled, subscription callbacks use blocking HTTP requests instead of async fire-and-forget. This ensures callbacks complete before the request handler returns, which is important for Lambda/serverless where async tasks may be lost when the function freezes after returning a response.

Callbacks are async (fire-and-forget) unless this method is called: the underlying setting defaults to False, so calling this method at all is what opts in.

Parameters:

enable – If True, use synchronous callbacks. Default is True, so calling with_sync_callbacks() with no arguments enables it.

Returns:

Self for method chaining.

with_thread_pool_workers(workers: int) → ActingWebApp[source]

Configure thread pool size for FastAPI integration.

The thread pool is used to execute synchronous ActingWeb handlers (database operations, HTTP requests) without blocking the async event loop.

Tuning guidelines: - Default: 10 workers (suitable for most applications) - Low traffic: 5 workers (reduces memory overhead) - High traffic: 20-50 workers (handles more concurrent requests) - Lambda: 5-10 workers (limited by function concurrency) - Container: Scale based on CPU cores (e.g., 2-5 per core)

Memory overhead: ~8MB per worker thread on average.

Parameters:

workers – Number of thread pool workers. Must be between 1 and 100.

Returns:

Self for method chaining.

Raises:

ValueError – If workers is outside the valid range [1, 100].

Example

>>> app = ActingWebApp(...).with_thread_pool_workers(20)
with_unique_creator(enable: bool = True) → ActingWebApp[source]

Enable unique creator constraint.

with_web_ui(enable: bool = True) → ActingWebApp[source]

Enable or disable the web UI.

ActorInterface

Improved Actor interface that wraps the core Actor class.

Provides a clean, intuitive interface for working with ActingWeb actors.

class actingweb.interface.actor_interface.ActorInterface(core_actor: Actor, service_registry=None, hooks: HookRegistry | None = None)[source]

Bases: object

Clean interface for ActingWeb actors.

This class wraps the core Actor class and provides a more intuitive interface for developers.

Example usage:

# Create new actor
actor = ActorInterface.create(
    creator="user@example.com",
    config=config,
)

# Access properties
actor.properties.email = "user@example.com"
actor.properties["settings"] = {"theme": "dark"}

# Manage trust relationships
peer = actor.trust.create_relationship(
    peer_url="https://peer.example.com/actor123",
    relationship="friend",
)

# Handle subscriptions
actor.subscriptions.subscribe_to_peer(
    peer_id="peer123",
    target="properties",
)

# Notify subscribers
actor.subscriptions.notify_subscribers(
    target="properties",
    data={"status": "active"},
)
as_client(client_id: str, trust_relationship: dict[str, Any] | None = None) → AuthenticatedActorView[source]

Create a view of this actor as seen by an OAuth2/MCP client.

All operations on this view will have permission checks enforced based on the client’s trust relationship.

Parameters:

  • client_id – The OAuth2/MCP client ID

  • trust_relationship – Optional trust relationship data

Returns:

AuthenticatedActorView with permission enforcement

Example

client_view = actor.as_client(“mcp_client_123”, trust_data) client_view.properties[“user_data”] = value # Permission checked

as_peer(peer_id: str, trust_relationship: dict[str, Any] | None = None) → AuthenticatedActorView[source]

Create a view of this actor as seen by a peer.

All operations on this view will have permission checks enforced based on the peer’s trust relationship.

Parameters:

  • peer_id – The peer actor’s ID

  • trust_relationship – Optional trust relationship data

Returns:

AuthenticatedActorView with permission enforcement

Example

peer_view = actor.as_peer(“peer123”, trust_data) peer_view.properties[“shared_data”] = value # Permission checked

property config

Get the ActingWeb configuration object.

Returns:

ActingWeb configuration instance

Raises:

RuntimeError – If config is not available

property core_actor: Actor

Access to underlying core actor (for advanced use).

classmethod create(creator: str, config: Config, actor_id: str | None = None, passphrase: str | None = None, delete_existing: bool = False, trustee_root: str | None = None, hooks: Any = None, service_registry=None) → ActorInterface[source]

Create a new actor.

Parameters:

  • creator – Creator identifier (usually email)

  • config – ActingWeb Config object

  • actor_id – Optional custom actor ID

  • passphrase – Optional custom passphrase

  • delete_existing – Whether to delete existing actor with same creator

  • trustee_root – Optional trustee root URL to set on the actor

  • hooks – Optional hook registry for executing lifecycle hooks

  • service_registry – Optional service registry for third-party service access

Returns:

New ActorInterface instance

property creator: str | None

Actor creator.

delete() → None[source]

Delete this actor and all associated data.

Note that this does not run lifecycle hooks — neither actor_deleted nor actor_deleted_complete. Only the HTTP DELETE /<actor_id> path fires those, so an application deleting an actor programmatically owns its own cleanup. A deletion tombstone is written either way (see get_deletion_status()).

classmethod get_by_creator(creator: str, config: Config, service_registry=None) → ActorInterface | None[source]

Get an existing actor by creator.

Parameters:

  • creator – Creator identifier

  • config – ActingWeb Config object

  • service_registry – Optional service registry for third-party service access

Returns:

ActorInterface instance or None if not found

classmethod get_by_id(actor_id: str, config: Config, service_registry=None) → ActorInterface | None[source]

Get an existing actor by ID.

Parameters:

  • actor_id – Actor ID

  • config – ActingWeb Config object

  • service_registry – Optional service registry for third-party service access

Returns:

ActorInterface instance or None if not found

classmethod get_by_property(property_name: str, property_value: str, config: Config, service_registry=None) → ActorInterface | None[source]

Get an existing actor by property value.

Parameters:

  • property_name – Property name to search

  • property_value – Property value to match

  • config – ActingWeb Config object

  • service_registry – Optional service registry for third-party service access

Returns:

ActorInterface instance or None if not found

classmethod get_deletion_status(actor_id: str, config: Config) → DeletionStatus[source]

Whether an actor has been deleted — or whether that cannot be determined.

Prefer this over get_by_id(...) is None for any guard of the form “do not do this work if the actor is gone”. Two reasons, both of which make the absence check wrong rather than merely weaker:

  • get_by_id() keeps returning an actor for the whole of delete(), because the actor row is removed last. The absence check fails open exactly in the window it exists for.

  • get_by_id() returns None for a deleted actor and for a failed read. A guard that skips on None also skips on a DynamoDB throttle — silently. For a paid-subscription webhook that means the customer paid and never got access.

This call answers with three values instead of two, from a tombstone written before the wipe begins. Treat UNKNOWN as “proceed”: the worst case is one orphan row an operator sweep can find, whereas treating it as “deleted” drops real work.

Costs one strongly-consistent point read.

from actingweb.interface import ActorInterface, DeletionStatus

status = ActorInterface.get_deletion_status(actor_id, config)
if status == DeletionStatus.DELETED:
    return  # late provider callback for a deleted account
# NOT_DELETED and UNKNOWN both proceed
Parameters:

  • actor_id – Actor ID

  • config – ActingWeb Config object

Returns:

DeletionStatus

See also

actingweb.deletion for the full rationale, and the actor_deleted_complete lifecycle hook for cleanup that must run after the wipe rather than racing it.

get_peer_info(peer_url: str) → dict[str, Any][source]

Get information about a peer actor.

Parameters:

peer_url – URL of the peer actor

Returns:

Dictionary with peer information

property id: str | None

Actor ID.

is_owner() → bool[source]

Check if current user is the owner of this actor.

is_valid() → bool[source]

Check if this actor is valid (has ID and exists).

modify_creator(new_creator: str) → bool[source]

Modify the creator of this actor.

Parameters:

new_creator – New creator identifier

Returns:

True if successful, False otherwise

property passphrase: str | None

Actor passphrase.

property properties: PropertyStore

Actor properties.

property property_lists

Actor property lists for distributed storage with subscription notifications.

refresh() → bool[source]

Refresh actor data from storage.

property services

Third-party service client manager.

property subscriptions: SubscriptionManager

Subscription manager.

to_dict() → dict[str, Any][source]

Convert actor to dictionary representation.

Returns:

Dictionary with actor data

property trust: TrustManager

Trust relationship manager.

property url: str

Actor URL.

Property Lists

Simplified property store interface for ActingWeb actors.

class actingweb.interface.property_store.NotifyingListProperty(list_prop: Any, list_name: str, actor: CoreActor | None = None)[source]

Bases: object

Wrapper around ListProperty that registers diffs for subscription notifications.

All list mutation operations will trigger register_diffs to notify subscribers of changes to the property list.

append(item: Any) → None[source]
clear() → None[source]
compact() → dict[str, Any][source]

Repair holes/orphans – see ListProperty.compact(). Registers a “metadata” diff (the closed operation vocabulary has no dedicated “compact” entry) so subscribers re-read the list rather than trust a positional diff for what is a storage-layer rewrite.

count(value: Any) → int[source]
delete() → None[source]
delete_by_handle(handle: ListItemHandle) → bool[source]

Single-shot value-addressed delete – see ListProperty.delete_by_handle(). Registers a “remove” diff (the closed operation vocabulary has no dedicated handle-delete entry) carrying the removed item’s decoded value, only when the delete actually succeeded – a False return means nothing changed.

extend(items: list[Any]) → None[source]
find(identity_key: str, value: Any, consistent: bool = True) → Any[source]

Read-only – see ListProperty.find().

find_all(identity_key: str, value: Any, consistent: bool = True) → list[Any][source]

Read-only – see ListProperty.find_all().

get_description() → str[source]
get_explanation() → str[source]
get_metadata() → dict[str, Any][source]

List metadata – see ListProperty.get_metadata(). Under v2, length is the advisory count_hint, not a counted value.

index(value: Any, start: int = 0, stop: int | None = None) → int[source]
insert(index: int, item: Any) → None[source]
items_with_handles() → list[tuple[ListItemHandle, Any]][source]

Read-only – see ListProperty.items_with_handles(). v2 only.

migrate_to_v2(allow_damaged: bool = False) → dict[str, Any][source]

Migrate this list from v1 to v2 storage – see ListProperty.migrate_to_v2(), including what allow_damaged gives up. Registers a “metadata” diff (same rationale as compact()) only when a migration actually happened.

pop(index: int = -1) → Any[source]
prime_from_rows(rows: dict[str, Any]) → None[source]
remove(value: Any) → None[source]
remove_where(identity_key: str, value: Any, *, first_only: bool = False) → int[source]

See ListProperty.remove_where(). Registers one “remove” diff per item actually removed, carrying its decoded value – the closed operation vocabulary has no dedicated bulk-remove entry, so a multi-match remove_where() looks to subscribers like several individual remove() calls.

ListProperty.remove_where() returns the removed items themselves (not just a count), so the diffs raised here name exactly the rows this call removed – no separate before-mutation snapshot, so nothing to drift out of sync with a concurrent mutation between two reads.

set_description(description: str) → None[source]
set_explanation(explanation: str) → None[source]
slice(start: int, end: int, consistent: bool = True) → list[Any][source]
to_indexed_list(consistent: bool = True) → list[tuple[int, Any]][source]
to_list(consistent: bool = True) → list[Any][source]
to_list_from_rows(rows: dict[str, Any]) → list[Any][source]
update_by_handle(handle: ListItemHandle, item: Any) → bool[source]

Single-shot value-addressed update – see ListProperty.update_by_handle(). Registers an “update” diff carrying both the new item and old_item (decoded from the handle), only when the update actually succeeded. No index – a handle is not positional.

update_where(identity_key: str, value: Any, item: Any, *, first_only: bool = False) → int[source]

See ListProperty.update_where(). Registers one “update” diff per item actually updated, carrying the new value and old_item (the pre-update value) – no index; a value-addressed update under v1 or v2 alike is not positional, and _apply_list_operation on the receiving side resolves an “update” diff by old_item when present regardless.

ListProperty.update_where() returns the pre-update values of the rows it actually updated, so this raises diffs from that directly instead of a separately-taken before-mutation snapshot – same rationale as remove_where() above.

verify(identity_key: str | None = None) → dict[str, Any][source]

Read-only integrity check – see ListProperty.verify(). No diff is registered; nothing changes.

Pass identity_key if your items carry an identifying field: duplicate detection defaults to byte comparison, which stops finding a duplicate once either copy is edited.

class actingweb.interface.property_store.PropertyListStore(core_list_store: Any, actor: CoreActor | None = None)[source]

Bases: object

Property list store wrapper that adds register_diffs for subscription notifications.

Wraps the core PropertyListStore and returns NotifyingListProperty instances for all list accesses, ensuring that list mutations trigger subscription notifications.

exists(name: str) → bool[source]

Check if a list property exists.

list_all() → list[str][source]

List all existing list property names.

list_all_with_rows() → tuple[list[str], dict[str, str]][source]

List all existing list property names, alongside the raw rows the names were derived from – see property.PropertyListStore.list_all_with_rows().

The rows are an opaque, point-in-time snapshot of the actor’s whole partition: feed them to a list’s prime_from_rows() / to_list_from_rows(), never inspect or parse a row name.

For ONE namespace of lists see list_prefix_with_rows(). Its cost does not go the way the names suggest: on a measured account this dump was 1,361.0 RCU over 11 queries and the five scoped reads covering the same lists were 1,363.5 over 15, so replacing one dump with several scoped calls is marginally worse. Note also that this method swallows a backend fault to ([], {}) while list_prefix_with_rows() raises – see there for why.

list_prefix_with_rows(prefix: str) → tuple[list[str], dict[str, str]][source]

List property names beginning with prefix, and their rows – see property.PropertyListStore.list_prefix_with_rows() for the full contract.

BOTH halves are scoped: names holds only the matching lists. Code migrating from list_all_with_rows() that keeps iterating names silently stops seeing every list outside the prefix.

prefix is a prefix, not a namespace – it also matches a list named exactly prefix and siblings like {prefix}-old. Pass the delimiter (“memory_”, not “memory”) if you mean a namespace.

Reads are eventually consistent; a backend fault RAISES DbError rather than returning empty, because for a scoped read an empty result is an ordinary answer. An empty prefix raises ValueError.

class actingweb.interface.property_store.PropertyStore(core_store: PropertyStore, actor: CoreActor | None = None, hooks: HookRegistry | None = None, config: Config | None = None)[source]

Bases: object

Clean interface for actor property management.

Provides dictionary-like access to actor properties with automatic subscription notifications and hook execution.

Example usage:

actor.properties.email = “user@example.com” actor.properties[“config”] = {“theme”: “dark”}

if “email” in actor.properties:

print(actor.properties.email)

for key, value in actor.properties.items():

print(f”{key}: {value}”)

clear() → None[source]

Clear all properties with diff registration.

property core_store: PropertyStore

Access underlying core property store.

delete(key: str) → bool[source]

Delete property and return True if it existed.

get(key: str, default: Any = None) → Any[source]

Get property value with default.

items() → Iterator[tuple[str, Any]][source]

Get all property key-value pairs (single bulk read).

keys() → Iterator[str][source]

Get all property keys.

set(key: str, value: Any) → None[source]

Set property value with hooks and diff registration.

set_without_notification(key: str, value: Any) → None[source]

Set property value without triggering subscription notifications.

Use this for internal operations where notifications are not desired.

to_dict() → dict[str, Any][source]

Convert to dictionary with one backend read.

Reads everything via get_all() instead of re-fetching each key individually (the old path issued one GetItem per property right after having bulk-read them all).

update(other: dict[str, Any]) → None[source]

Update properties from dictionary with hooks and diff registration.

values() → Iterator[Any][source]

Get all property values (single bulk read).

Hooks Registry

Hook system for ActingWeb applications.

Provides a clean decorator-based system for registering hooks that respond to various ActingWeb events.

class actingweb.interface.hooks.HookMetadata(description: str = '', input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None)[source]

Bases: object

Metadata for method/action hooks.

This metadata is used to describe hooks for API discovery via GET /<actor_id>/methods and GET /<actor_id>/actions endpoints.

description

Human-readable description of what the hook does

Type:

str

input_schema

JSON schema describing expected input parameters

Type:

dict[str, Any] | None

output_schema

JSON schema describing the expected return value. This describes the ActingWeb /actions / /methods discovery surface only. It is NOT forwarded to MCP: a tool’s outputSchema comes solely from @mcp_tool(output_schema=...). And neither one causes structuredContent to be emitted – only a structuredContent key in the returned dict does that.

Type:

dict[str, Any] | None

annotations

Safety/behavior hints (e.g., readOnlyHint, destructiveHint)

Type:

dict[str, Any] | None

annotations: dict[str, Any] | None = None
description: str = ''
input_schema: dict[str, Any] | None = None
output_schema: dict[str, Any] | None = None
class actingweb.interface.hooks.HookRegistry[source]

Bases: object

Registry for managing application hooks.

Hooks allow applications to customize ActingWeb behavior at key points without modifying the core library.

execute_action_hooks(action_name: str, actor: Any, data: Any, auth_context: dict[str, Any] | None = None) → Any[source]

Execute action hooks with transparent permission checking.

Note: If you have async hooks and are in an async context, use execute_action_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_action_hooks_async(action_name: str, actor: Any, data: Any, auth_context: dict[str, Any] | None = None) → Any[source]

Execute action hooks with native async support.

See execute_method_hooks_async for details.

Parameters:

  • action_name – Name of the action hook to execute

  • actor – ActorInterface instance

  • data – Request data/parameters

  • auth_context – Optional authentication context

Returns:

Result from the first successful hook, or None

execute_app_callback_hooks(callback_name: str, data: Any) → bool | dict[str, Any][source]

Execute application-level callback hooks (no actor context).

Note: If you have async hooks and are in an async context, use execute_app_callback_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_app_callback_hooks_async(callback_name: str, data: Any) → bool | dict[str, Any][source]

Execute application-level callback hooks asynchronously (no actor context).

Parameters:

  • callback_name – Name of the callback

  • data – Callback data

Returns:

True if processed, or result data dict

execute_callback_hooks(callback_name: str, actor: Any, data: Any) → bool | dict[str, Any][source]

Execute callback hooks and return whether callback was processed or result data.

Note: If you have async hooks and are in an async context, use execute_callback_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_callback_hooks_async(callback_name: str, actor: Any, data: Any) → bool | dict[str, Any][source]

Execute callback hooks asynchronously.

Parameters:

  • callback_name – Name of the callback

  • actor – ActorInterface instance

  • data – Callback data

Returns:

True if processed, or result data dict

execute_lifecycle_hooks(event: str, actor: Any, **kwargs: Any) → Any[source]

Execute lifecycle hooks.

Note: If you have async hooks and are in an async context, use execute_lifecycle_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_lifecycle_hooks_async(event: str, actor: Any, **kwargs: Any) → Any[source]

Execute lifecycle hooks asynchronously.

Parameters:

  • event – Lifecycle event name

  • actor – ActorInterface instance

  • **kwargs – Additional event-specific arguments

Returns:

Result from the last hook, or None

execute_method_hooks(method_name: str, actor: Any, data: Any, auth_context: dict[str, Any] | None = None) → Any[source]

Execute method hooks with transparent permission checking.

Note: If you have async hooks and are in an async context, use execute_method_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_method_hooks_async(method_name: str, actor: Any, data: Any, auth_context: dict[str, Any] | None = None) → Any[source]

Execute method hooks with native async support.

Use this method when calling from an async context (FastAPI handlers). Supports both sync and async hooks: - Async hooks are awaited directly - Sync hooks are called directly (sync-compatible)

Parameters:

  • method_name – Name of the method hook to execute

  • actor – ActorInterface instance

  • data – Request data/parameters

  • auth_context – Optional authentication context

Returns:

Result from the first successful hook, or None

execute_property_hooks(property_name: str, operation: str, actor: Any, value: Any, path: list[str] | None = None, auth_context: dict[str, Any] | None = None) → Any[source]

Execute property hooks with transparent permission checking.

Note: If you have async hooks and are in an async context, use execute_property_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_property_hooks_async(property_name: str, operation: str, actor: Any, value: Any, path: list[str] | None = None, auth_context: dict[str, Any] | None = None) → Any[source]

Execute property hooks asynchronously with transparent permission checking.

Parameters:

  • property_name – Name of the property

  • operation – Operation being performed (get, put, post, delete)

  • actor – ActorInterface instance

  • value – Property value

  • path – Optional path components for nested properties

  • auth_context – Optional authentication context

Returns:

Modified property value or None if operation was rejected

execute_subscription_hooks(actor: Any, subscription: dict[str, Any], peer_id: str, data: Any) → bool[source]

Execute subscription hooks and return whether subscription was processed.

Note: If you have async hooks and are in an async context, use execute_subscription_hooks_async() instead for proper async execution. Async hooks in this method will be executed via asyncio.run() which may cause issues if already in an event loop.

async execute_subscription_hooks_async(actor: Any, subscription: dict[str, Any], peer_id: str, data: Any) → bool[source]

Execute subscription hooks asynchronously.

Parameters:

  • actor – ActorInterface instance

  • subscription – Subscription information

  • peer_id – ID of the peer sending the subscription event

  • data – Event data

Returns:

True if subscription was processed

get_action_metadata_list() → list[dict[str, Any]][source]

Get list of all registered actions with their metadata.

Returns a list of dictionaries containing name and metadata for each action. Wildcard (*) hooks are excluded from listing.

Returns:

List of dicts with name, description, input_schema, output_schema, annotations

get_method_metadata_list() → list[dict[str, Any]][source]

Get list of all registered methods with their metadata.

Returns a list of dictionaries containing name and metadata for each method. Wildcard (*) hooks are excluded from listing.

Returns:

List of dicts with name, description, input_schema, output_schema, annotations

register_action_hook(action_name: str, func: Callable[[...], Any]) → None[source]

Register an action hook function.

Parameters:

  • action_name – Name of action to hook (“*” for all actions)

  • func – Function with signature (actor, action_name, data) -> Any

register_app_callback_hook(callback_name: str, func: Callable[[...], Any]) → None[source]

Register an application-level callback hook function.

Parameters:

  • callback_name – Name of callback to hook (e.g., “bot”, “oauth”)

  • func – Function with signature (data) -> Any (no actor parameter)

register_callback_hook(callback_name: str, func: Callable[[...], Any]) → None[source]

Register a callback hook function.

Parameters:

  • callback_name – Name of callback to hook (“*” for all callbacks)

  • func – Function with signature (actor, name, data) -> bool

register_lifecycle_hook(event: str, func: Callable[[...], Any]) → None[source]

Register a lifecycle hook function.

Parameters:

  • event – Lifecycle event name

  • func – Function with signature (actor, **kwargs) -> Any

register_method_hook(method_name: str, func: Callable[[...], Any]) → None[source]

Register a method hook function.

Parameters:

  • method_name – Name of method to hook (“*” for all methods)

  • func – Function with signature (actor, method_name, data) -> Any

register_property_hook(property_name: str, func: Callable[[...], Any]) → None[source]

Register a property hook function.

Parameters:

  • property_name – Name of property to hook (“*” for all properties)

  • func – Function with signature (actor, operation, value, path) -> Any

register_subscription_hook(func: Callable[[...], Any]) → None[source]

Register a subscription hook function.

Parameters:

func – Function with signature (actor, subscription, peer_id, data) -> bool

class actingweb.interface.hooks.HookType(*values)[source]

Bases: Enum

Types of hooks available.

ACTION = 'action'
CALLBACK = 'callback'
LIFECYCLE = 'lifecycle'
METHOD = 'method'
PROPERTY = 'property'
SUBSCRIPTION = 'subscription'
class actingweb.interface.hooks.LifecycleEvent(*values)[source]

Bases: Enum

Lifecycle events that can be hooked.

ACTOR_CREATED = 'actor_created'
ACTOR_DELETED = 'actor_deleted'
ACTOR_DELETED_COMPLETE = 'actor_deleted_complete'
OAUTH_SUCCESS = 'oauth_success'
SUBSCRIPTION_DELETED = 'subscription_deleted'
TRUST_APPROVED = 'trust_approved'
TRUST_DELETED = 'trust_deleted'
class actingweb.interface.hooks.PropertyOperation(*values)[source]

Bases: Enum

Property operations that can be hooked.

DELETE = 'delete'
GET = 'get'
POST = 'post'
PUT = 'put'
actingweb.interface.hooks.action_hook(action_name: str = '*', description: str = '', input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None) → Callable[[...], Any][source]

Decorator for registering action hooks with optional metadata.

Parameters:

  • action_name – Name of action to hook (“*” for all actions)

  • description – Human-readable description of what the action does

  • input_schema – JSON schema describing expected input parameters

  • output_schema – JSON schema describing the expected return value. This describes the ActingWeb /actions / /methods discovery surface only. It is NOT forwarded to MCP: a tool’s outputSchema comes solely from @mcp_tool(output_schema=...). And neither one causes structuredContent to be emitted – only a structuredContent key in the returned dict does that.

  • annotations – Safety/behavior hints (e.g., destructiveHint, readOnlyHint)

Example

@action_hook(
    "delete_record",
    description="Permanently delete a record",
    input_schema={
        "type": "object",
        "properties": {"record_id": {"type": "string"}},
        "required": ["record_id"]
    },
    annotations={"destructiveHint": True, "readOnlyHint": False}
)
def handle_delete(actor, action_name, data):
    # Execute trigger-based action
    delete_record(data.get("record_id"))
    return {"status": "deleted"}
actingweb.interface.hooks.app_callback_hook(callback_name: str) → Callable[[...], Any][source]

Decorator for registering application-level callback hooks (no actor context).

Parameters:

callback_name – Name of callback to hook (e.g., “bot”, “oauth”)

Example

@app_callback_hook("bot")
def handle_bot_callback(data):
    # Process bot callback (no actor context)
    return True
actingweb.interface.hooks.callback_hook(callback_name: str = '*') → Callable[[...], Any][source]

Decorator for registering actor-level callback hooks.

Parameters:

callback_name – Name of callback to hook (“*” for all)

Example

@callback_hook("ping")
def handle_ping_callback(actor, name, data):
    # Process actor-level callback
    return True
actingweb.interface.hooks.get_hook_metadata(func: Callable[[...], Any]) → HookMetadata[source]

Get hook metadata from a decorated function.

Priority order: 1. Explicit _hook_metadata (from decorator parameters) 2. MCP metadata (_mcp_metadata from @mcp_tool decorator) 3. Auto-generated schemas from TypedDict type hints

For each source, if input_schema or output_schema is not provided, attempts to auto-generate from function type hints (TypedDict only).

Parameters:

func – The hook function to get metadata from

Returns:

HookMetadata instance with the function’s metadata

actingweb.interface.hooks.get_hook_registry() → HookRegistry[source]

Get the global hook registry.

actingweb.interface.hooks.lifecycle_hook(event: str) → Callable[[...], Any][source]

Decorator for registering lifecycle hooks.

Parameters:

event – Lifecycle event name

Example

@lifecycle_hook("actor_created")
def on_actor_created(actor, **kwargs):
    # Initialize actor
    actor.properties.created_at = datetime.now()
actingweb.interface.hooks.method_hook(method_name: str = '*', description: str = '', input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None) → Callable[[...], Any][source]

Decorator for registering method hooks with optional metadata.

Parameters:

  • method_name – Name of method to hook (“*” for all methods)

  • description – Human-readable description of what the method does

  • input_schema – JSON schema describing expected input parameters

  • output_schema – JSON schema describing the expected return value. This describes the ActingWeb /actions / /methods discovery surface only. It is NOT forwarded to MCP: a tool’s outputSchema comes solely from @mcp_tool(output_schema=...). And neither one causes structuredContent to be emitted – only a structuredContent key in the returned dict does that.

  • annotations – Safety/behavior hints (e.g., readOnlyHint, idempotentHint)

Example

@method_hook(
    "calculate",
    description="Perform a mathematical calculation",
    input_schema={
        "type": "object",
        "properties": {"x": {"type": "number"}},
        "required": ["x"]
    },
    annotations={"readOnlyHint": True}
)
def handle_calculate_method(actor, method_name, data):
    # Execute RPC-style method
    result = perform_calculation(data)
    return {"result": result}
actingweb.interface.hooks.property_hook(property_name: str = '*', operations: list[str] | None = None) → Callable[[...], Any][source]

Decorator for registering property hooks.

Parameters:

  • property_name – Name of property to hook (“*” for all)

  • operations – List of operations to hook (default: all)

Example

@property_hook("email", ["get", "put"])
def handle_email(actor, operation, value, path):
    if operation == "get":
        return value if actor.is_owner() else None
    elif operation == "put":
        return value.lower() if "@" in value else None
    return value
actingweb.interface.hooks.subscription_hook(func: Callable[[...], Any]) → Callable[source]

Decorator for registering subscription hooks.

Example

@subscription_hook
def handle_subscription(actor, subscription, peer_id, data):
    # Process subscription callback
    return True

Integrations

Runtime Context

Runtime Context System for ActingWeb.

This module provides a generic system for making request-specific context available to hook functions during request processing. This solves the architectural constraint where hook functions have fixed signatures but need access to request-specific context.

Architecture Problem:

  • ActingWeb hook functions have fixed signatures: hook(actor, action_name, data)

  • Multiple clients can access the same actor (MCP clients, web users, API clients)

  • Each request needs context about the current client/request for proper handling

  • Can’t modify hook signatures without breaking framework compatibility

  • The actor object handed to hooks is frequently the same cached object reused across many requests (MCP’s per-actor cache is the primary example), so context cannot live as mutable state on the actor without leaking between requests and, under concurrency, between callers.

Solution:

  • Store context in a module-level contextvars.ContextVar, keyed by actor id, rather than as an attribute on the actor object.

  • ContextVar is task-local under asyncio and thread-local under a plain WSGI worker thread, so context set for one request is invisible to a concurrently running request even when both share the same actor object.

  • Provide type-safe access methods with clear documentation

  • Support multiple context types (MCP, OAuth2, web sessions, etc.)

  • Framework integrations (Flask, FastAPI) clear the ContextVar at the end of each request, via the module-level clear_all_context(), so nothing survives onto a reused worker thread. clear_all_context() is framework/test plumbing – application hooks should use RuntimeContext(actor).clear_context() if they need to clear a specific actor’s context mid-request.

Usage Example:

# During request authentication:
runtime_context = RuntimeContext(actor)
runtime_context.set_mcp_context(
    client_id="mcp_abc123",
    trust_relationship=trust_obj,
    peer_id="oauth2_client:user@example.com:mcp_abc123"
)

# In hook functions:
def handle_search(actor, action_name, data):
    runtime_context = RuntimeContext(actor)
    mcp_context = runtime_context.get_mcp_context()
    if mcp_context:
        client_name = mcp_context.trust_relationship.client_name
        # Customize behavior based on client type

Concurrency notes:

  • A request dispatched onto a fresh asyncio task automatically inherits the context that was active when the task was created (a snapshot, not a live link — the child cannot mutate the parent’s view and vice versa).

  • A request dispatched onto a new thread (e.g. ThreadPoolExecutor) does not automatically inherit the caller’s context; the submitting code must capture contextvars.copy_context() and run the callable via ctx.run(...) for context to cross the hop. See actingweb/interface/hooks.py for the one place ActingWeb does this.

class actingweb.runtime_context.MCPContext(client_id: str, trust_relationship: Any, peer_id: str, token_data: dict[str, Any] | None = None, transport_session_id: str | None = None, client_info: dict[str, Any] | None = None)[source]

Bases: object

Runtime context for MCP (Model Context Protocol) requests.

Contains information about the current MCP client making the request, allowing tools to customize behavior based on client capabilities.

client_id: str
client_info: dict[str, Any] | None = None
peer_id: str
token_data: dict[str, Any] | None = None
transport_session_id: str | None = None
trust_relationship: Any
class actingweb.runtime_context.OAuth2Context(client_id: str, user_email: str, scopes: list[str], token_data: dict[str, Any] | None = None)[source]

Bases: object

Runtime context for OAuth2 authenticated requests.

Contains information about the current OAuth2 session for web or API access.

client_id: str
scopes: list[str]
token_data: dict[str, Any] | None = None
user_email: str
class actingweb.runtime_context.RuntimeContext(actor: Any)[source]

Bases: object

Generic runtime context manager for ActingWeb actors.

Provides type-safe access to request-specific context, stored in a ContextVar keyed by actor id rather than as state on the actor object. This solves the architectural constraint where hook functions can’t receive additional parameters, without letting context leak across requests that happen to share a cached actor object.

The context is request-scoped (task-local under asyncio, thread-local under a plain worker thread) and is cleared by the framework integration at the end of each request.

clear_context() → None[source]

Clear all runtime context for this actor.

This should be called after request processing is complete to avoid context leaking between requests.

get_custom_context(key: str) → Any[source]

Get custom context data.

Parameters:

key – Context key

Returns:

Context value or None if not found

get_mcp_context() → MCPContext | None[source]

Get MCP context for the current request.

Returns:

MCPContext if this is an MCP request, None otherwise

get_oauth2_context() → OAuth2Context | None[source]

Get OAuth2 context for the current request.

Returns:

OAuth2Context if this is an OAuth2 request, None otherwise

get_request_type() → str | None[source]

Determine the type of the current request.

Returns:

“mcp”, “oauth2”, “web”, or None if no context is set

get_web_context() → WebContext | None[source]

Get web browser context for the current request.

Returns:

WebContext if this is a web request, None otherwise

has_context() → bool[source]

Check if any runtime context is set.

Returns:

True if any context is set for this actor

set_custom_context(key: str, value: Any) → None[source]

Set custom context data.

Parameters:

  • key – Context key (avoid ‘mcp’, ‘oauth2’, ‘web’ which are reserved)

  • value – Context value

set_mcp_context(client_id: str, trust_relationship: Any, peer_id: str, token_data: dict[str, Any] | None = None, transport_session_id: str | None = None, client_info: dict[str, Any] | None = None) → None[source]

Set MCP context for the current request.

Parameters:

  • client_id – OAuth2 client ID of the MCP client

  • trust_relationship – Trust database record with client metadata

  • peer_id – Normalized peer identifier for permission checking

  • token_data – Optional OAuth2 token metadata

  • transport_session_id – Optional per-MCP-connection identifier that distinguishes concurrent sessions sharing one OAuth2 credential (see MCPContext.transport_session_id).

  • client_info – Optional live clientInfo dict from the active session’s initialize call (see MCPContext.client_info).

set_oauth2_context(client_id: str, user_email: str, scopes: list[str], token_data: dict[str, Any] | None = None) → None[source]

Set OAuth2 context for the current request.

Parameters:

  • client_id – OAuth2 client ID

  • user_email – Authenticated user email

  • scopes – Granted OAuth2 scopes

  • token_data – Optional token metadata

set_web_context(session_id: str | None = None, user_agent: str | None = None, ip_address: str | None = None, authenticated_user: str | None = None) → None[source]

Set web browser context for the current request.

Parameters:

  • session_id – Session identifier

  • user_agent – Browser user agent

  • ip_address – Client IP address

  • authenticated_user – Authenticated user identifier

class actingweb.runtime_context.WebContext(session_id: str | None = None, user_agent: str | None = None, ip_address: str | None = None, authenticated_user: str | None = None)[source]

Bases: object

Runtime context for web browser requests.

Contains session and authentication information for web UI access.

authenticated_user: str | None = None
ip_address: str | None = None
session_id: str | None = None
user_agent: str | None = None
actingweb.runtime_context.clear_all_context() → None[source]

Clear runtime context for every actor tracked in the calling thread’s/task’s own context.

Framework integrations call this once at the end of a request/response cycle (Flask teardown_request, FastAPI middleware finally) as a blanket reset, since a request handler does not always know in advance which actor id(s) it touched. Test fixtures use it to reset state between tests that share a thread or event loop.

This resets only the calling thread’s/task’s own view of _runtime_context_var – it has no effect on a concurrently running request’s context, which lives in its own copied Context.

actingweb.runtime_context.get_client_info_from_context(actor: Any) → dict[str, str] | None[source]

Helper function to extract client information from runtime context.

This provides a unified way to get client details regardless of context type.

Parameters:

actor – Actor object with potential runtime context

Returns:

Dict with ‘name’, ‘version’, ‘platform’ keys, or None if no client info available

Authentication

actingweb.auth.check_and_verify_auth(appreq=None, actor_id=None, config=None)[source]

Check and verify authentication for non-ActingWeb routes.

This function provides authentication verification for custom routes that don’t go through the standard ActingWeb handler system. It performs authentication checks and is designed for use in custom application routes.

Parameters:

  • appreq – Request object in the format used by ActingWeb handlers.

  • actor_id (str | None) – Actor ID to verify authentication against.

  • config (Config | None) – ActingWeb config object.

Returns:

A dictionary with the following keys:

  • authenticated (bool): True if authentication successful.

  • actor (Actor | None): Actor object when authenticated, otherwise None.

  • auth (Auth): Auth object with authentication details.

  • response (dict): Response details: {"code": int, "text": str, "headers": dict}.

  • redirect (str | None): Redirect URL if authentication requires redirect.

Return type:

dict

Example

auth_result = check_and_verify_auth(appreq, actor_id, config)
if not auth_result['authenticated']:
    if auth_result['response']['code'] == 302:
        # Redirect for OAuth
        return redirect(auth_result['redirect'])
    # Return error response
    return error_response(
        auth_result['response']['code'], auth_result['response']['text']
    )

# Authentication successful, use auth_result['actor']
actor = auth_result['actor']