actingweb code

Subpackages

• actingweb.db.dynamodb package
• actingweb.handlers package

Submodules

actingweb.actor module

class actingweb.actor.Actor(actor_id: str | None = None, config: Any | None = None)

Bases: object

callback_subscription(peerid=None, sub_obj=None, sub=None, diff=None, blob=None)

async callback_subscription_async(peerid=None, sub_obj=None, sub=None, diff=None, blob=None)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

create(url: str, creator: str, passphrase: str, actor_id: str | None = None, delete: bool = False, trustee_root: str | None = None, hooks: Any = None) → bool

“Creates a new actor and persists it.

If delete is True, any existing actors with same creator value will be deleted. If it is False, the one with the correct passphrase will be chosen (if any)

create_reciprocal_trust(url, secret=None, desc='', relationship='', trust_type='')

Creates a new reciprocal trust relationship locally and by requesting a relationship from a peer actor.

Parameters:

• relationship – The trust type/permission level to request (friend, admin, etc.)

• trust_type – Expected peer mini-app type for validation (optional)

async create_reciprocal_trust_async(url, secret=None, desc='', relationship='', trust_type='')

Async version of create_reciprocal_trust - prevents blocking on peer HTTP calls.

Creates a new reciprocal trust relationship locally and by requesting a relationship from a peer actor.

create_remote_subscription(peerid=None, target=None, subtarget=None, resource=None, granularity=None)

Creates a new subscription at peerid.

async create_remote_subscription_async(peerid=None, target=None, subtarget=None, resource=None, granularity=None)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

create_subscription(peerid=None, target=None, subtarget=None, resource=None, granularity=None, subid=None, callback=False)

create_verified_trust(baseuri='', peerid=None, approved=False, secret=None, verification_token=None, trust_type=None, peer_approved=None, relationship=None, desc='')

Creates a new trust when requested and call backs to initiating actor to verify relationship.

Parameters:

• trust_type – The peer’s ActingWeb mini-application type URI

• relationship – The trust type/permission level (friend, admin, etc.)

async create_verified_trust_async(baseuri='', peerid=None, approved=False, secret=None, verification_token=None, trust_type=None, peer_approved=None, relationship=None, desc='')

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

delete() → None

Deletes an actor and cleans up all relevant stored data

delete_peer_trustee(shorttype=None, peerid=None)

delete_properties()

Deletes all properties.

delete_property(name)

Deletes a property name. (DEPRECATED, use actor’s property store!)

delete_reciprocal_trust(peerid=None, delete_peer=False)

Deletes a trust relationship and requests deletion of peer’s relationship as well.

async delete_reciprocal_trust_async(peerid=None, delete_peer=False)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

delete_remote_subscription(peerid=None, subid=None)

async delete_remote_subscription_async(peerid=None, subid=None)

Async version - wraps sync method in asyncio.to_thread to prevent blocking.

delete_subscription(peerid=None, subid=None, callback=False)

Deletes a specified subscription

get(actor_id: str | None = None) → dict[str, Any] | None

Retrieves an actor from storage or initialises if it does not exist

get_from_creator(creator: str | None = None) → bool

Initialise an actor by matching on creator/email.

Returns True if an actor could be loaded, otherwise False. When multiple actors share the same creator (possible when unique_creator is disabled), the first deterministic match will be selected in order to provide stable behaviour for login flows that do not specify an explicit actor ID.

get_from_property(name: str = 'oauthId', value: str | None = None) → None

Initialise an actor by matching on a stored property.

Use with caution as the property’s value de-facto becomes a security token. If multiple properties are found with the same value, no actor will be initialised. Also note that this is a costly operation as all properties of this type will be retrieved and proceessed.

get_peer_info(url: str, max_retries: int = 3, retry_delay: float = 0.5) → dict[str, Any]

Contacts another actor over http/s to retrieve meta information.

Includes retry logic for transient network failures with exponential backoff.

Note: This sync method blocks the event loop. In FastAPI/uvicorn contexts, use AsyncTrustHandler which calls create_reciprocal_trust_async() instead.

Parameters:

• url – Root URI of a remote actor

• max_retries – Maximum number of retry attempts (default: 3)

• retry_delay – Initial delay between retries in seconds (default: 0.5)

Return type:

dict

Returns:

The json response from the /meta path in the data element and last_response_code/last_response_message set to the results of the https request

Example:

{
    "last_response_code": 200,
    "last_response_message": "OK",
    "data": {}
}

async get_peer_info_async(url: str) → dict[str, Any]

Async version of get_peer_info using httpx.

Contacts another actor over HTTP/S to retrieve meta information without blocking.

Parameters:

url – Root URI of a remote actor

Returns:

Dict with last_response_code, last_response_message, and data

get_peer_trustee(shorttype=None, peerid=None)

Get a peer, either existing or create it as trustee

Will retrieve an existing peer or create a new and establish trust. If no trust exists, a new trust will be established. Use either peerid to target a specific known peer, or shorttype to allow creation of a new peer if none exists

get_properties()

Retrieves properties from db and returns a dict.

get_property(name)

Retrieves a property object named name. (DEPRECATED, use actor’s property store!)

get_subscription(peerid=None, subid=None, callback=False)

Retrieves a single subscription identified by peerid and subid.

get_subscription_obj(peerid=None, subid=None, callback=False)

Retrieves a single subscription identified by peerid and subid.

get_subscriptions(peerid: str | None = None, target: str | None = None, subtarget: str | None = None, resource: str | None = None, callback: bool | None = None) → list[dict[str, Any]] | None

Retrieves subscriptions from db.

Parameters:

• peerid – Filter by peer ID (None = all peers)

• target – Filter by target (None = all targets)

• subtarget – Filter by subtarget (None = all subtargets)

• resource – Filter by resource (None = all resources)

• callback – Filter by callback flag (None = all, False = inbound, True = outbound)

Returns:

List of subscription dictionaries, or None if actor has no ID

get_trust_relationship(peerid=None)

get_trust_relationships(relationship='', peerid='', trust_type='')

Retrieves all trust relationships or filtered.

is_subscription_suspended(target: str, subtarget: str | None = None) → bool

Check if diff registration is suspended for a target/subtarget.

Parameters:

• target – Target resource (e.g., “properties”)

• subtarget – Optional subtarget (e.g., property name)

Returns:

True if suspended, False otherwise

modify(creator: str | None = None) → bool

modify_trust_and_notify(relationship=None, peerid=None, baseuri='', secret='', desc='', approved=None, verified=None, verification_token=None, peer_approved=None, client_name=None, client_version=None, client_platform=None, oauth_client_id=None, last_accessed=None, last_connected_via=None)

Changes a trust relationship and noties the peer if approval is changed.

async modify_trust_and_notify_async(relationship=None, peerid=None, baseuri='', secret='', desc='', approved=None, verified=None, verification_token=None, peer_approved=None, client_name=None, client_version=None, client_platform=None, oauth_client_id=None, last_accessed=None, last_connected_via=None)

Async version of modify_trust_and_notify - prevents blocking on peer notification.

Changes a trust relationship and notifies the peer if approval is changed. Database operations remain synchronous, but peer HTTP notification is async.

register_diffs(target=None, subtarget=None, resource=None, blob=None)

Registers a blob diff against all subscriptions with the correct target, subtarget, and resource.

If resource is set, the blob is expected to be the FULL resource object, not a diff.

Note: Skips registration if the target/subtarget is currently suspended. Use suspend_subscriptions() and resume_subscriptions() to manage suspension.

resume_subscriptions(target: str, subtarget: str | None = None) → int

Resume diff registration and send resync callbacks.

Sends a resync callback to ALL subscriptions on this target/subtarget, telling them to do a full GET to re-sync their state.

Parameters:

• target – Target resource (e.g., “properties”)

• subtarget – Optional subtarget (e.g., property name)

Returns:

The number of resync callbacks sent

set_property(name, value)

Sets an actor’s property name to value. (DEPRECATED, use actor’s property store!)

suspend_subscriptions(target: str, subtarget: str | None = None) → bool

Suspend diff registration for a target/subtarget.

While suspended, property changes will NOT register diffs or trigger callbacks. Call resume_subscriptions() to lift suspension and send resync callbacks.

Parameters:

• target – Target resource (e.g., “properties”)

• subtarget – Optional subtarget (e.g., property name)

Returns:

True if newly suspended, False if already suspended

exception actingweb.actor.ActorError

Bases: Exception

Base exception class for Actor-related errors.

exception actingweb.actor.ActorNotFoundError

Bases: ActorError

Raised when an actor cannot be found.

class actingweb.actor.Actors(config=None)

Bases: object

Handles all actors

fetch()

class actingweb.actor.DummyPropertyClass(v: Any = None)

Bases: object

Only used to deprecate get_property() in 2.4.4

exception actingweb.actor.InvalidActorDataError

Bases: ActorError

Raised when actor data is invalid or corrupted.

exception actingweb.actor.PeerCommunicationError

Bases: ActorError

Raised when communication with peer actors fails.

exception actingweb.actor.TrustRelationshipError

Bases: ActorError

Raised when trust relationship operations fail.

actingweb.attribute module

class actingweb.attribute.Attributes(actor_id: str | None = None, bucket: str | None = None, config: Any | None = None)

Bases: object

Attributes is the main entity keeping an attribute.

It needs to be initalized at object creation time.

conditional_update_attr(name: str | None = None, old_data: Any | None = None, new_data: Any | None = None, timestamp: Any | None = None) → bool

Conditionally update an attribute only if current data matches old_data.

This provides atomic compare-and-swap functionality for race-free updates.

Parameters:

• name – Attribute name

• old_data – Expected current data value (for comparison)

• new_data – New data to set if current matches old_data

• timestamp – Optional timestamp

Returns:

True if update succeeded (current matched old_data), False otherwise

delete_attr(name: str | None = None) → bool

delete_bucket() → bool

Deletes the attribute bucket in the database

get_attr(name: str | None = None) → dict[str, Any] | None

Retrieves a single attribute

get_bucket() → dict[str, Any] | None

Retrieves the attribute bucket from the database

set_attr(name: str | None = None, data: Any | None = None, timestamp: Any | None = None, ttl_seconds: int | None = None) → bool

Sets new data for this attribute.

Parameters:

• name – Attribute name

• data – Data to store (JSON-serializable)

• timestamp – Optional timestamp

• ttl_seconds – Optional TTL in seconds. If provided, DynamoDB will automatically delete this item after expiry.

class actingweb.attribute.Buckets(actor_id: str | None = None, config: Any | None = None)

Bases: object

Handles all attribute buckets of a specific actor_id

Access the attributes in .props as a dictionary

delete() → bool

fetch() → dict[str, dict[str, dict[str, Any]]] | bool

fetch_timestamps() → dict[str, Any] | bool

class actingweb.attribute.InternalStore(actor_id: str | None = None, config: Any | None = None, bucket: str | None = None)

Bases: object

Access to internal attributes using .prop notation

actingweb.auth module

class actingweb.auth.Auth(actor_id, auth_type='basic', config=None)

Bases: object

The auth class handles authentication and authorisation for the various schemes supported.

The check_authentication() function checks the various authentication schemes against the path and does proper authentication. There are two types supported: basic (using creator credentials) and token (received when trust is created or OAuth2). The check_authorisation() function validates the authenticated user against the config.py access list. check_token_auth() can be called from outside the class to do a simple peer/bearer token verification.

The response[], acl[], and authn_done variables are useful outside Auth(). authn_done is set when authentication has been done and a final authentication status can be found in response[].

self.response = {

“code”: 403, # Result code (http) “text”: “Forbidden”, # Proposed response text “headers”: [], # Headers to add to response after authentication has been done

}

self.acl = {

“authenticated”: False, # Has authentication been verified and passed? “authorised”: False, # Has authorisation been done and appropriate acls set? “rights”: ‘’, # “a”, “r” (approve or reject) “relationship”: None, # E.g. creator, friend, admin, etc “peerid”: ‘’, # Peerid if there is a relationship “approved”: False, # True if the peer is approved

}

check_authentication(appreq, path)

Checks authentication in appreq, redirecting back to path if oauth is done.

check_authorisation(path='', subpath='', method='', peerid='', approved=True)

Checks if the authenticated user has acl access rights in config.py.

Takes the path, subpath, method, and peerid of the path (if auth user is different from the peer that owns the path, e.g. creator). If approved is False, then the trust relationship does not need to be approved for access

check_token_auth(appreq, via_hint: str | None = None)

Validate bearer tokens and optionally record how the connection occurred.

async check_token_auth_async(appreq, via_hint: str | None = None)

Async version: Validate bearer tokens without blocking the event loop.

actingweb.auth.add_auth_response(appreq=None, auth_obj=None)

Called after authentication to set appropriate HTTP response based on auth result.

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

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']

async actingweb.auth.check_and_verify_auth_async(appreq=None, actor_id=None, config=None)

Async version: Check and verify authentication for non-ActingWeb routes.

This async function provides authentication verification for custom routes that don’t go through the standard ActingWeb handler system. It uses async HTTP calls to avoid blocking the event loop during OAuth2 token validation.

Use this in async FastAPI endpoints instead of check_and_verify_auth() when the endpoint might receive OAuth2 tokens that need validation against the provider.

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 = await check_and_verify_auth_async(appreq, actor_id, config)
if not auth_result['authenticated']:
    if auth_result['response']['code'] == 302:
        return RedirectResponse(auth_result['redirect'])
    return JSONResponse(
        status_code=auth_result['response']['code'],
        content={"error": auth_result['response']['text']}
    )

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

actingweb.aw_proxy module

class actingweb.aw_proxy.AwProxy(trust_target: Any = None, peer_target: dict[str, Any] | None = None, config: Any = None, timeout: int | float | tuple[int | float, int | float] | None = None)

Bases: object

Proxy to other trust peers to execute RPC style calls.

Initialise with either trust_target to target a specific existing trust or use peer_target for simplicity to use the trust established with the peer.

Parameters:

• trust_target – Trust object for the target peer

• peer_target – Simplified peer target dict

• config – Configuration object

• timeout – HTTP timeout in seconds. Either a single value (used for both connect and read timeouts) or a tuple (connect_timeout, read_timeout). Default: (5, 20) = 5s connect, 20s read timeout.

Provides both sync methods (using requests) and async methods (using httpx) for peer communication:

• Sync: get_resource(), create_resource(), change_resource(), delete_resource()

• Async: get_resource_async(), create_resource_async(), change_resource_async(), delete_resource_async()

Use async methods in FastAPI routes for non-blocking I/O.

change_resource(path=None, params=None)

async change_resource_async(path: str | None = None, params: dict[str, Any] | None = None) → dict[str, Any] | None

Async version of change_resource (PUT) using httpx.

Parameters:

• path – The resource path on the peer actor

• params – Data to send as JSON body

Returns:

The JSON response from the peer, or None if the request failed.

create_resource(path=None, params=None)

async create_resource_async(path: str | None = None, params: dict[str, Any] | None = None) → dict[str, Any] | None

Async version of create_resource (POST) using httpx.

Parameters:

• path – The resource path on the peer actor

• params – Data to send as JSON body

Returns:

The JSON response from the peer, or None if the request failed.

delete_resource(path=None)

async delete_resource_async(path: str | None = None) → dict[str, Any] | None

Async version of delete_resource (DELETE) using httpx.

Parameters:

path – The resource path on the peer actor

Returns:

The JSON response from the peer, or None if the request failed.

get_resource(path=None, params=None)

async get_resource_async(path: str | None = None, params: dict[str, Any] | None = None) → dict[str, Any] | None

Async version of get_resource using httpx.

Use this method in async contexts (e.g., FastAPI routes) for non-blocking HTTP calls to peer actors.

Parameters:

• path – The resource path on the peer actor (e.g., “trust/friend/permissions”)

• params – Optional query parameters

Returns:

The JSON response from the peer, or None if the request failed.

actingweb.aw_web_request module

class actingweb.aw_web_request.AWRequest(url: str | None = None, params: dict[str, Any] | None = None, body: str | None = None, headers: dict[str, str] | None = None, cookies: dict[str, str] | None = None)

Bases: object

arguments() → list[str]

get(var: str = '') → str

get_header(header: str = '') → str

class actingweb.aw_web_request.AWResponse

Bases: object

set_cookie(name: str, value: str, max_age: int = 1209600, path: str = '/', secure: bool = True, httponly: bool = False, samesite: str = 'Lax') → None

set_redirect(url: str) → None

set_status(code: int = 200, message: str = 'Ok') → bool | None

write(body: str | None = None, encode: bool = False) → bool | None

class actingweb.aw_web_request.AWWebObj(url: str | None = None, params: dict[str, Any] | None = None, body: str | None = None, headers: dict[str, str] | None = None, cookies: dict[str, str] | None = None)

Bases: object

actingweb.config module

class actingweb.config.Config(**kwargs: Any)

Bases: object

DbActor: ModuleType

DbAttribute: ModuleType

DbPeerTrustee: ModuleType

DbProperty: ModuleType

DbSubscription: ModuleType

DbSubscriptionDiff: ModuleType

DbSubscriptionSuspension: ModuleType

DbTrust: ModuleType

static new_token(length: int = 40) → str

static new_uuid(seed: str) → str

update_supported_options() → None

Update aw_supported based on enabled features.

This method is called when features are dynamically enabled after Config initialization (e.g., via ActingWebApp builder methods).

actingweb.peertrustee module

class actingweb.peertrustee.PeerTrustee(actor_id=None, peerid=None, short_type=None, peer_type=None, config=None)

Bases: object

create(baseuri=None, passphrase=None)

delete()

get() → dict[str, Any] | bool | None

Retrieve peer trustee from database.

Returns:

Peer trustee data if exactly one match found False: When multiple peer trustees of same type exist (ambiguous lookup) None: When no peer trustee found or required parameters missing

Return type:

dict[str, Any]

actingweb.property module

class actingweb.property.Properties(actor_id: str | None = None, config: Any | None = None)

Bases: object

Handles all properties of a specific actor_id

Access the properties in .props as a dictionary

delete() → bool

fetch() → dict[str, Any] | bool

class actingweb.property.Property(actor_id: str | None = None, name: str | None = None, value: Any | None = None, config: Any | None = None)

Bases: object

property is the main entity keeping a property.

It needs to be initalised at object creation time.

delete() → bool | None

Deletes the property in the database

get() → Any

Retrieves the property from the database

get_actor_id() → str | None

set(value: Any) → bool

Sets a new value for this property

class actingweb.property.PropertyListStore(actor_id: str | None = None, config: Any | None = None)

Bases: object

Explicit interface for managing list properties.

Used when the application knows it’s working with list data.

exists(name: str) → bool

Check if a list property exists without creating it.

list_all() → list[str]

List all existing list property names.

class actingweb.property.PropertyStore(actor_id: str | None = None, config: Any | None = None)

Bases: object

get_all() → dict[str, Any]

Fetch all properties from the database and return as dictionary.

actingweb.subscription module

class actingweb.subscription.Subscription(actor_id=None, peerid=None, subid=None, callback=False, config=None)

Bases: object

Base class with core subscription methods (storage-related)

add_diff(blob=None)

Add a new diff for this subscription

clear_diff(seqnr)

Clears one specific diff

clear_diffs(seqnr=0)

Clear all diffs up to and including a seqnr

create(target: str | None = None, subtarget: str | None = None, resource: str | None = None, granularity: str | None = None, seqnr: int = 0) → bool

Create new subscription and push it to db

decrease_seq()

Rollback sequence number by 1 (used when diff creation fails after seq increment)

delete()

Delete a subscription in storage

get() → dict[str, Any]

Retrieve subscription from db given pre-initialized variables

get_diff(seqnr=0)

Get one specific diff

get_diffs()

Get all the diffs available for this subscription ordered by the timestamp, oldest first

increase_seq()

class actingweb.subscription.Subscriptions(actor_id=None, config=None)

Bases: object

Handles all subscriptions of a specific actor_id

Access the indvidual subscriptions in .dbsubscriptions and the subscription data in .subscriptions as a dictionary

delete()

fetch()

actingweb.trust module

class actingweb.trust.Trust(actor_id: str | None = None, peerid: str | None = None, token: str | None = None, config: Any | None = None)

Bases: object

create(baseuri: str = '', peer_type: str = '', relationship: str = '', secret: str = '', approved: bool = False, verified: bool = False, verification_token: str = '', desc: str = '', peer_approved: bool = False, peer_identifier: str | None = None, established_via: str | None = None, created_at: str | None = None, last_accessed: str | None = None, last_connected_via: str | None = None) → bool

Create a new trust relationship

delete() → bool

Delete the trust relationship

get() → dict[str, Any] | None

Retrieve a trust relationship with either peerid or token

modify(baseuri: str | None = None, secret: str | None = None, desc: str | None = None, approved: bool | None = None, verified: bool | None = None, verification_token: str | None = None, peer_approved: bool | None = None, peer_identifier: str | None = None, established_via: str | None = None, created_at: str | None = None, last_accessed: str | None = None, last_connected_via: str | None = None, client_name: str | None = None, client_version: str | None = None, client_platform: str | None = None, oauth_client_id: str | None = None, aw_supported: str | None = None, aw_version: str | None = None, capabilities_fetched_at: str | None = None) → bool

class actingweb.trust.Trusts(actor_id: str | None = None, config: Any | None = None)

Bases: object

Handles all trusts of a specific actor_id

Access the indvidual trusts in .dbtrusts and the trust data in .trusts as a dictionary

delete() → bool

fetch() → list[dict[str, Any]] | list[Any] | None

actingweb.trust.canonical_connection_method(method: str | None) → str | None

Normalize connection hints to canonical channels.

Parameters:

method – Connection method hint (e.g., “oauth2:interactive”, “trust”, “mcp”)

Returns:

Canonical connection method or original value if not recognized

Examples

• “oauth2:interactive” -> “oauth”

• “OAUTH” -> “oauth”

• “trust” -> “trust”

• “subscription” -> “subscription”

• “mcp” -> “mcp”

• “unknown_method” -> “unknown_method”

actingweb.runtime_context module

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)

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)

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)

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

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

Get custom context data.

Parameters:

key – Context key

Returns:

Context value or None if not found

get_mcp_context() → MCPContext | None

Get MCP context for the current request.

Returns:

MCPContext if this is an MCP request, None otherwise

get_oauth2_context() → OAuth2Context | None

Get OAuth2 context for the current request.

Returns:

OAuth2Context if this is an OAuth2 request, None otherwise

get_request_type() → str | None

Determine the type of the current request.

Returns:

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

get_web_context() → WebContext | None

Get web browser context for the current request.

Returns:

WebContext if this is a web request, None otherwise

has_context() → bool

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

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

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

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

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)

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

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

actingweb.request_context module

Request context management for logging correlation.

This module provides thread-safe and async-safe storage for request-scoped context information (request ID, actor ID, peer ID) using Python’s contextvars.

The context is automatically isolated per request and propagates correctly through async/await boundaries, making it suitable for both Flask (WSGI) and FastAPI (ASGI) applications.

actingweb.request_context.clear_request_context() → None

Clear all request context values.

This should be called at the end of request handling to prevent context leakage between requests. Framework integrations should call this in finally blocks or response handlers.

Example

>>> set_request_context(actor_id="actor123")
>>> get_actor_id()
'actor123'
>>> clear_request_context()
>>> get_actor_id()  # Returns None

actingweb.request_context.format_context_compact() → str

Format context as a compact string for text-based logging.

Format: [short_request_id:actor_id:short_peer_id] Missing values are represented as “-”

Returns:

actor123:peer456]”

Return type:

Formatted context string like “[a1b2c3d4

Example

>>> set_request_context(
...     request_id="550e8400-e29b-41d4-a716-446655440000",
...     actor_id="actor123",
... )
>>> set_peer_id("urn:actingweb:example.com:peer456")
>>> format_context_compact()
'[55440000:actor123:peer456]'
>>> clear_request_context()
>>> format_context_compact()
'[-:-:-]'

actingweb.request_context.generate_request_id() → str

Generate a new UUID4 request ID.

Returns:

“550e8400-e29b-41d4-a716-446655440000”

Return type:

A string UUID in the format

Example

>>> request_id = generate_request_id()
>>> len(request_id)
36

actingweb.request_context.get_actor_id() → str | None

Get the actor ID for the current context.

Returns:

The actor ID if set, otherwise None

Example

>>> get_actor_id()  # Returns None if not set
>>> set_actor_id("actor123")
>>> get_actor_id()
'actor123'

actingweb.request_context.get_context_dict() → dict[str, Any]

Get all context values as a dictionary.

This is useful for structured logging (JSON) where context should be included as separate fields rather than in a formatted string.

Returns:

request_id, actor_id, peer_id Values are None if not set in current context

Return type:

Dictionary with keys

Example

>>> set_request_context(
...     request_id="550e8400-e29b-41d4-a716-446655440000",
...     actor_id="actor123",
... )
>>> get_context_dict()
{'request_id': '550e8400-e29b-41d4-a716-446655440000', 'actor_id': 'actor123', 'peer_id': None}

actingweb.request_context.get_peer_id() → str | None

Get the peer ID for the current context.

Returns:

The peer ID if set, otherwise None

Example

>>> get_peer_id()  # Returns None if not set
>>> set_peer_id("peer456")
>>> get_peer_id()
'peer456'

actingweb.request_context.get_request_id() → str | None

Get the request ID for the current context.

Returns:

The request ID if set, otherwise None

Example

>>> get_request_id()  # Returns None if not set
>>> set_request_id("550e8400-e29b-41d4-a716-446655440000")
>>> get_request_id()
'550e8400-e29b-41d4-a716-446655440000'

actingweb.request_context.get_short_peer_id() → str

Get a shortened version of the peer ID (last segment after final colon).

Peer IDs often follow the pattern “urn:actingweb:example.com:actor123”, so this extracts just the last segment for compact logging.

Returns:

Last segment of peer ID after final colon, or “-” if no peer ID is set

Example

>>> set_peer_id("urn:actingweb:example.com:actor123")
>>> get_short_peer_id()
'actor123'
>>> set_peer_id("simple_peer")
>>> get_short_peer_id()
'simple_peer'
>>> set_peer_id(None)
>>> get_short_peer_id()
'-'

actingweb.request_context.get_short_request_id() → str

Get a shortened version of the request ID (last 8 characters).

This is useful for compact log formats where the full UUID is too verbose. 8 characters provides sufficient uniqueness for correlation within a reasonable time window.

Returns:

Last 8 characters of the request ID, or “-” if no request ID is set

Example

>>> set_request_id("550e8400-e29b-41d4-a716-446655440000")
>>> get_short_request_id()
'40000'
>>> set_request_id(None)
>>> get_short_request_id()
'-'

actingweb.request_context.set_actor_id(actor_id: str | None) → None

Set the actor ID for the current context.

Parameters:

actor_id – The actor ID to set, or None to clear

Example

>>> set_actor_id("actor123")
>>> get_actor_id()
'actor123'

actingweb.request_context.set_peer_id(peer_id: str | None) → None

Set the peer ID for the current context.

The peer ID identifies the remote actor or client making the request, and is typically set after authentication completes.

Parameters:

peer_id – The peer ID to set, or None to clear

Example

>>> set_peer_id("peer456")
>>> get_peer_id()
'peer456'

actingweb.request_context.set_request_context(request_id: str | None = None, actor_id: str | None = None, peer_id: str | None = None, *, generate_id: bool = True) → str

Set all request context values at once.

This is the primary entry point for framework integrations (Flask, FastAPI) to establish request context at the start of request handling.

Parameters:

• request_id – The request ID, or None to generate a new one

• actor_id – The actor ID from the request path

• peer_id – The peer ID (typically set later after authentication)

• generate_id – If True and request_id is None, generate a new UUID

Returns:

The request ID that was set (either provided or generated)

Example

>>> # Framework integration: set context at request start
>>> req_id = set_request_context(
...     request_id=request.headers.get("X-Request-ID"),
...     actor_id=request.path_params.get("actor_id"),
...     generate_id=True,
... )
>>> print(f"Handling request {req_id}")

actingweb.request_context.set_request_id(request_id: str | None) → None

Set the request ID for the current context.

Parameters:

request_id – The request ID to set, or None to clear

Example

>>> set_request_id("550e8400-e29b-41d4-a716-446655440000")
>>> get_request_id()
'550e8400-e29b-41d4-a716-446655440000'

actingweb.logging_config module

Centralized logging configuration for ActingWeb.

This module provides utilities for configuring logging across the ActingWeb framework with sensible defaults for different environments.

actingweb.logging_config.configure_actingweb_logging(level: int = 20, *, db_level: int | None = None, handlers_level: int | None = None, interface_level: int | None = None, oauth_level: int | None = None, mcp_level: int | None = None) → None

Configure ActingWeb logging with sensible defaults.

This function sets up hierarchical logging for different ActingWeb subsystems, allowing fine-grained control over verbosity in different parts of the framework.

Parameters:

• level – Default level for all actingweb loggers (default: INFO)

• db_level – Override for database operations (default: WARNING to reduce noise)

• handlers_level – Override for HTTP handlers (default: uses main level)

• interface_level – Override for interface layer (default: uses main level)

• oauth_level – Override for OAuth2 components (default: uses main level)

• mcp_level – Override for MCP protocol (default: uses main level)

Example

Development setup (verbose):

>>> import logging
>>> from actingweb.logging_config import configure_actingweb_logging
>>> configure_actingweb_logging(logging.DEBUG)

Production setup (quiet DB, normal handlers):

>>> configure_actingweb_logging(
...     level=logging.WARNING,
...     handlers_level=logging.INFO,
...     db_level=logging.ERROR,
... )

Testing setup (only errors):

>>> configure_actingweb_logging(logging.ERROR)

actingweb.logging_config.configure_actingweb_logging_with_context(level: int = 20, *, db_level: int | None = None, handlers_level: int | None = None, interface_level: int | None = None, oauth_level: int | None = None, mcp_level: int | None = None, enable_context: bool = True, structured: bool = False) → None

Configure ActingWeb logging with request context support.

This is a convenience function that combines configure_actingweb_logging() with enable_request_context_filter() and format configuration.

Parameters:

• level – Default level for all actingweb loggers (default: INFO)

• db_level – Override for database operations (default: WARNING)

• handlers_level – Override for HTTP handlers (default: uses main level)

• interface_level – Override for interface layer (default: uses main level)

• oauth_level – Override for OAuth2 components (default: uses main level)

• mcp_level – Override for MCP protocol (default: uses main level)

• enable_context – If True, enable request context injection (default: True)

• structured – If True, use structured context for JSON logging (default: False)

Example

Development with context:

>>> import logging
>>> configure_actingweb_logging_with_context(logging.DEBUG)

Production with context:

>>> configure_actingweb_logging_with_context(
...     level=logging.WARNING,
...     handlers_level=logging.INFO,
...     db_level=logging.ERROR,
... )

Disable context:

>>> configure_actingweb_logging_with_context(
...     level=logging.INFO,
...     enable_context=False,
... )

actingweb.logging_config.configure_development_logging(*, verbose: bool = False) → None

Configure logging for development environments.

Parameters:

verbose – If True, enable DEBUG logging everywhere (default: False)

Example

>>> configure_development_logging()  # INFO level
>>> configure_development_logging(verbose=True)  # DEBUG level

actingweb.logging_config.configure_production_logging(*, http_traffic: bool = True, lifecycle_events: bool = True) → None

Configure logging for production environments with performance focus.

This is an opinionated production configuration that balances observability with performance.

Parameters:

• http_traffic – If True, log HTTP request handling at INFO level

• lifecycle_events – If True, log actor lifecycle events at INFO level

Example

>>> configure_production_logging()
>>> # HTTP traffic and lifecycle events logged, everything else quiet

actingweb.logging_config.configure_testing_logging(*, debug: bool = False) → None

Configure logging for test environments.

By default, only shows errors during tests to keep output clean. Can be overridden with debug=True for troubleshooting.

Parameters:

debug – If True, show all DEBUG logs (default: False)

Example

>>> import os
>>> # Enable debug logging with environment variable
>>> debug_tests = os.getenv("ACTINGWEB_DEBUG") == "1"
>>> configure_testing_logging(debug=debug_tests)

actingweb.logging_config.enable_request_context_filter(*, logger: str | Logger = 'actingweb', structured: bool = False, handler_type: Literal['all', 'stream', 'file'] = 'all') → None

Enable request context injection for ActingWeb loggers.

This function adds RequestContextFilter to existing logging handlers, enabling automatic injection of request context (request ID, actor ID, peer ID) into log records.

After enabling the filter, you should update log formats to include the %(context)s placeholder:

formatter = logging.Formatter(get_context_format()) handler.setFormatter(formatter)

Parameters:

• logger – Logger name or Logger object to add filters to (default: “actingweb”)

• structured – If True, use StructuredContextFilter for JSON logging; if False, use RequestContextFilter for text logging

• handler_type – Which handlers to add filter to: - “all”: Add to all handlers - “stream”: Add only to StreamHandler instances - “file”: Add only to FileHandler instances

Example

Text logging with context:

>>> import logging
>>> from actingweb.logging_config import (
...     configure_actingweb_logging,
...     enable_request_context_filter,
...     get_context_format,
... )
>>> configure_actingweb_logging(logging.INFO)
>>> enable_request_context_filter()
>>> # Update format to include context
>>> for handler in logging.getLogger("actingweb").handlers:
...     formatter = logging.Formatter(get_context_format())
...     handler.setFormatter(formatter)

JSON logging with structured context:

>>> enable_request_context_filter(structured=True)
>>> # Use custom JSON formatter that accesses record.request_id, etc.

actingweb.logging_config.get_context_format(*, include_timestamp: bool = True, include_context: bool = False, include_logger: bool = True, include_level: bool = True) → str

Generate a log format string with optional request context.

This function creates format strings suitable for use with logging.Formatter. When include_context is True, the format includes a %(context)s placeholder that will be populated by RequestContextFilter.

IMPORTANT: If include_context=True, you MUST add RequestContextFilter to your handlers using enable_request_context_filter() or add_context_filter_to_handler(), otherwise formatting will fail with “Formatting field not found in record: ‘context’”.

Parameters:

• include_timestamp – Include timestamp in format (default: True)

• include_context – Include request context placeholder (default: False)

• include_logger – Include logger name (default: True)

• include_level – Include log level (default: True)

Returns:

A format string suitable for logging.Formatter

Example

>>> get_context_format()
'%(asctime)s %(name)s:%(levelname)s: %(message)s'
>>> get_context_format(include_context=True)
'%(asctime)s %(context)s %(name)s:%(levelname)s: %(message)s'
>>> get_context_format(include_timestamp=False, include_context=False)
'%(name)s:%(levelname)s: %(message)s'

actingweb.logging_config.get_performance_critical_loggers() → list[str]

Return list of loggers that should be WARNING+ in production.

These loggers are in hot paths and excessive logging impacts performance. Use this to configure production environments where performance is critical.

Returns:

List of logger names that are performance-sensitive

Example

>>> for logger_name in get_performance_critical_loggers():
...     logging.getLogger(logger_name).setLevel(logging.WARNING)

actingweb.log_filter module

Logging filters for request context injection.

This module provides logging.Filter classes that automatically inject request context (request ID, actor ID, peer ID) into log records, enabling correlation and filtering without modifying existing log statements.

class actingweb.log_filter.RequestContextFilter(name='')

Bases: Filter

Logging filter that injects request context into log records.

This filter adds a ‘context’ attribute to each LogRecord containing formatted context information. The context includes: - Request ID (short form, last 8 chars) - Actor ID (full) - Peer ID (short form, last segment after colon)

The context is formatted as: [req_id:actor_id:peer_id] Missing values are represented as “-”

Usage:

>>> import logging
>>> from actingweb.log_filter import RequestContextFilter
>>> from actingweb.request_context import set_request_context
>>>
>>> # Set up logger with filter
>>> logger = logging.getLogger("myapp")
>>> handler = logging.StreamHandler()
>>> handler.addFilter(RequestContextFilter())
>>> formatter = logging.Formatter(
...     "%(asctime)s %(context)s %(name)s:%(levelname)s: %(message)s"
... )
>>> handler.setFormatter(formatter)
>>> logger.addHandler(handler)
>>>
>>> # Context automatically appears in logs
>>> set_request_context(actor_id="actor123")
>>> logger.info("Processing request")
# Output: 2024-01-15 10:23:45,123 [a1b2c3d4:actor123:-] myapp:INFO: Processing request

None - all state comes from contextvars in request_context module

filter(record: LogRecord) → bool

Add context to the log record.

This method is called by the logging framework for each log record. It adds a ‘context’ attribute containing the formatted context string.

Parameters:

record – The LogRecord to process

Returns:

Always True (record is never filtered out)

class actingweb.log_filter.StructuredContextFilter(name='')

Bases: Filter

Logging filter that injects request context as separate fields.

This filter is designed for structured logging (JSON) where context values should be separate fields rather than a formatted string.

The filter adds these attributes to each LogRecord: - request_id: Full request ID (UUID) - actor_id: Actor ID - peer_id: Peer ID

Values are None if not set in current context.

Usage:

>>> import logging
>>> import json
>>> from actingweb.log_filter import StructuredContextFilter
>>> from actingweb.request_context import set_request_context
>>>
>>> # Set up logger with filter for JSON output
>>> logger = logging.getLogger("myapp")
>>> handler = logging.StreamHandler()
>>> handler.addFilter(StructuredContextFilter())
>>>
>>> # Custom JSON formatter (simplified example)
>>> class JsonFormatter(logging.Formatter):
...     def format(self, record):
...         return json.dumps({
...             "timestamp": record.created,
...             "level": record.levelname,
...             "logger": record.name,
...             "message": record.getMessage(),
...             "request_id": getattr(record, "request_id", None),
...             "actor_id": getattr(record, "actor_id", None),
...             "peer_id": getattr(record, "peer_id", None),
...         })
>>>
>>> handler.setFormatter(JsonFormatter())
>>> logger.addHandler(handler)
>>>
>>> # Context automatically included in JSON
>>> set_request_context(actor_id="actor123")
>>> logger.info("Processing request")
# Output: {"timestamp": 1234567890.123, "level": "INFO", ...}

None - all state comes from contextvars in request_context module

filter(record: LogRecord) → bool

Add context fields to the log record.

This method is called by the logging framework for each log record. It adds separate attributes for each context field.

Parameters:

record – The LogRecord to process

Returns:

Always True (record is never filtered out)

actingweb.log_filter.add_context_filter_to_handler(handler: Handler, *, structured: bool = False) → None

Add a request context filter to a logging handler.

This is a convenience function for adding the appropriate filter type to an existing handler.

Parameters:

• handler – The logging handler to add the filter to

• structured – If True, use StructuredContextFilter for JSON logging; if False, use RequestContextFilter for text logging

Example

>>> import logging
>>> handler = logging.StreamHandler()
>>> add_context_filter_to_handler(handler)
>>> # Handler now has RequestContextFilter attached

actingweb.log_filter.add_context_filter_to_logger(logger: Logger | str, *, structured: bool = False) → None

Add a request context filter to all handlers of a logger.

This is a convenience function for adding filters to an existing logger’s handlers. It’s useful for adding context to loggers that are already configured.

Parameters:

• logger – The logger (or logger name) to add filters to

• structured – If True, use StructuredContextFilter for JSON logging; if False, use RequestContextFilter for text logging

Example

>>> import logging
>>> logging.basicConfig()  # Set up basic logging
>>> add_context_filter_to_logger("actingweb")
>>> # All actingweb loggers now include request context