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
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
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_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_email_as_creator(enable: bool = True) → ActingWebApp[source]: Force email property as creator.

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. Use environment variable INDEXED_PROPERTIES for runtime override.

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

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

When False (default), uses new lookup table approach which supports property values larger than 2048 bytes. When True, uses legacy DynamoDB GSI or PostgreSQL index on 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

Set this to True during migration from legacy systems. Once all properties are migrated to lookup table, set back to False (default).

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. 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_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.

Parameters:: enable – If True, use synchronous callbacks. Default is True.
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.

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

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.

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

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'

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
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
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 attaching runtime context to actor objects 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

Solution: - Attach runtime context to actor objects during request processing - Provide type-safe access methods with clear documentation - Support multiple context types (MCP, OAuth2, web sessions, etc.) - Clean up context after request completion

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

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 that gets attached to actor objects during request processing. This solves the architectural constraint where hook functions can’t receive additional parameters.

The context is request-scoped and should be cleaned up after processing.

clear_context() → None[source]

Clear all runtime context from the 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 attached to the 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.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']