SDK API Reference

Complete specifications for all modules, classes, and methods of agenticstar-platform SDK (v3.0.15).

Installation

Install the SDK using pip.

Basic Installationcurl

pip install agenticstar-platform==3.0.15

The basic install (no extras) depends on httpx / pydantic / pyyaml (pyyaml>=6.0 was added in v1.0.1+). If your base image or other packages pin PyYAML, make sure the pinned version is 6.0 or later.

To install only specific modules, specify extras.

Extra	説明
[[db]]	PostgreSQL database support
[[rag]]	RAG (embedding + vector search) support
[[storage]]	Cloud storage (Azure, S3, GCS) support
[[security]]	Security (content moderation, PII detection) support
[[auth]]	AGENTIC STAR Auth service integration
[[memory]]	Semantic Memory (Mem0 + Qdrant) support
[[all]]	All modules

Install Specific Modulescurl

# RAG module
pip install agenticstar-platform[rag]==3.0.15

# DB + RAG + Storage
pip install agenticstar-platform[db,rag,storage]==3.0.15

# Storage (per provider, v0.5.28+)
pip install agenticstar-platform[storage-aws]==3.0.15

# All modules
pip install agenticstar-platform[all]==3.0.15

---

Module List

The SDK consists of the following 13 modules.

Module	Description	Key Classes
events	Async event delivery system (SSE / Webhook / DB)	EventEmitter, StreamingEvent
db	PostgreSQL connection + Azure AD auth. Completion report for cycle executions (v3.0.14+)	PostgreSQLManager, DataAccess, report_completion
rag	RAG (embedding + vector search via Qdrant)	EmbeddingGenerator, QdrantManager
storage	Cloud storage (Azure, S3, GCS; per-provider extras: storage-azure / storage-aws / storage-gcp)	AzureBlobStorageClient, S3StorageClient
auth	AGENTIC STAR Auth API client	AgenticStarAuthClient
memory	Semantic memory (Mem0 + Qdrant)	SemanticMemoryClient
security	Content Moderation + PII Detection	AzureSecurityClient, ContentSafetyValidator
common	Common utilities (secret masking)	SecretMasker
metering	LLM usage & cost metering (ledger + daily rollup, v0.5.15+)	UsageMeter
runner	Marketplace-compatible execution lifecycle (identity check → fetch input → run → persist result/Webhook → terminal exactly once → cleanup, v0.5.30+)	run_marketplace_agent
audit	Action-audit ledger (append-only record of approvals / tool authorizations / policy violations; callers pass digests instead of raw content, v0.5.31+; alerts writers v0.5.32+)	ActionAudit, GatewayActionAudit, GuardrailAlertWriter, GuardrailAlertAggregates
workflow	Runner for agents that run in steps of a project workflow (mp-workflow/1; Marketplace edition 3.1 and later, v1.0.1+)	run, WorkflowContext, AuditApi
policy	Tool-call governance policy (pure computation that combines two layers and decides allow / forbid, v1.0.0+)	tighten_merge, evaluate, EffectivePolicy

---

events Events Module

Async event delivery system. Emit events via EventEmitter and deliver them to SSE / Webhook / DB.

Enums

EventType

Event types displayed in the frontend UI. For how to implement each member so it actually displays, see the implementation guide in the Events / Streaming Guide.

PHASE_START = "phase_start"

PROGRESS_UPDATE = "progress_update"

THOUGHT_MESSAGE = "thought_message"

COMPLETION_SUCCESS = "completion_success"

COMPLETION_FAILURE = "completion_failure"

USER_INTERACTION_REQUIRED = "user_interaction_required"

UNEXPECTED_ERROR = "unexpected_error"

HITL_REQUIRED_BROWSER_VNC = "hitl_required_browser_vnc"

HITL_REQUIRED_BROWSER_CLI = "hitl_required_browser_cli"

HITL_COMPLETED = "hitl_completed"

FILE_CREATED = "file_created"

PERMISSION_REQUEST = "permission_request"

PERMISSION_RESPONSE = "permission_response"

TOOL_START = "tool_start"

TOOL_RESULT = "tool_result"

PROGRESS_MESSAGE = "progress_message"

SubEventType

Sub-event types that classify an event more finely. sub_event_type alone does not change the rendering — specify metadata.actionType to vary how it displays.

SEARCH_WEB = "search_web"

COMMAND_EXECUTION = "command_execution"

FILE_OPERATION = "file_operation"

LOCAL_ASSISTANT = "local_assistant"

MCP_TOOL = "mcp_tool"

FILE_EDITED = "file_edited"

FILE_READ = "file_read"

FILE_SEARCHED = "file_searched"

BASH_EXECUTED = "bash_executed"

WEB_FETCHED = "web_fetched"

TASK_LAUNCHED = "task_launched"

TODO_UPDATED = "todo_updated"

VIDEO_GENERATED = "video_generated"

IMAGE_GENERATED = "image_generated"

SLIDE_CREATED = "slide_created"

MACOS_AUTOMATION = "macos_automation"

Data Classes

StreamingEvent

Event data for real-time SSE delivery.

@dataclass class StreamingEvent: event_type: EventType execution_id: str message: str timestamp: float metadata: Optional[Dict[str, Any]]= None sub_event_type: Optional[SubEventType]= None def to_dict() -> Dict[str, Any]: ...

SequencedEvent

Event with ordering guarantee.

@dataclass class SequencedEvent: sequence: int # >= 0 event_type: EventType execution_id: str message: str timestamp: float metadata: Optional[Dict[str, Any]]= None sub_event_type: Optional[SubEventType]= None def __post_init__(): ... # Validates sequence >= 0 def to_streaming_event() -> StreamingEvent: ... def to_dict() -> Dict[str, Any]: ...

ExecutionMessage

Database-stored message for marketplace UI integration.

@dataclass class ExecutionMessage: user_id: str conversation_id: str message_id: str chunk_type: str content_data: Dict[str, Any] def to_dict() -> Dict[str, Any]: ... @classmethod def from_streaming_event( event: StreamingEvent, user_id: str, conversation_id: str, message_id: str, chunk_type: Optional[str]= None ) -> ExecutionMessage: ...

Classes

EventEmitter

Async event queue. Emits and consumes events in a non-blocking manner.

class EventEmitter: def __init__( execution_id: str, handler: Optional[EventHandler]= None ): ... @property def is_completed() -> bool: ... @property def sequence_number() -> int: ... async def emit_event( event_type: EventType, message: str, metadata: Optional[Dict[str, Any]]= None, sub_event_type: Optional[SubEventType]= None ) -> None: ... async def emit( event_type: EventType, message: str, metadata: Optional[Dict[str, Any]]= None, sub_event_type: Optional[SubEventType]= None ) -> None: ... async def consume_events( timeout: float = 0.1 ) -> AsyncGenerator[str,None]: ... async def drain( timeout: float = 0.1 ) -> None: ... def mark_completed() -> None: ... async def cleanup() -> None: ...

Returns

emit_event() -> NoneAdds an event to the queue. If a handler is set, it is called asynchronously

emit() -> NoneAlias (shorthand) for emit_event()

consume_events() -> AsyncGenerator[str, None]Asynchronously yields SSE-formatted strings. Stops automatically on completion

drain() -> NoneDrains the queue by driving the registered handler (for use cases that do not yield SSE). Internally runs consume_events() and terminates automatically after the completion event

Emitting and Consuming EventsPython

from agenticstar_platform import EventEmitter, EventType

emitter = EventEmitter(execution_id="exec-abc-123")

# Emit event
await emitter.emit_event(
    event_type=EventType.PHASE_START,
    message="Starting document analysis",
    metadata={"phase": "analysis", "total_pages": 42}
)

# Consume events (SSE streaming)
async for chunk in emitter.consume_events():
    print(chunk)

DatabaseEventHandler

Persists events to the PostgreSQL execution_messages table.

class DatabaseEventHandler: def __init__( data_access: Any, user_id: str, conversation_id: str, message_id: str, table_name: str = "execution_messages", chunk_type_map: Optional[Dict[EventType, str]]= None ): ... async def __call__(event: StreamingEvent) -> Optional[str]: ...

WebhookEventHandler

Sends events to an HTTP webhook endpoint.

class WebhookEventHandler: def __init__( webhook_url: str, conversation_id: str, message_id: str, headers: Optional[Dict[str, str]]= None, timeout_seconds: int = 30, message_type: str = "append_message", token_provider: Optional[Callable[[], Optional[str]]]= None, chunk_type_map: Optional[Dict[EventType, str]]= None, request_source: Optional[str]= None ): ... async def __call__(event: StreamingEvent) -> Optional[str]: ...

CompositeEventHandler

Executes multiple EventHandlers in parallel.

class CompositeEventHandler: def __init__( handlers: List[Callable], return_first_chunk: bool = True ): ... async def __call__(event: StreamingEvent) -> Optional[str]: ...

Factory Functions

def create_sse_handler() -> EventHandler

Creates a handler in SSE (Server-Sent Events) format.

def create_json_handler() -> EventHandler

Creates a handler in JSON format.

def create_marketplace_handler( data_access: Any, webhook_url: str, user_id: str, conversation_id: str, message_id: str, token_provider: Optional[Callable[[], Optional[str]]]= None, request_source: Optional[str]= None ) -> CompositeEventHandler

Creates a composite handler (DB + Webhook) for marketplace UI.

Protocol

EventHandler

Generic interface for event processing.

class EventHandler(Protocol): async def __call__( self, event: StreamingEvent ) -> Optional[str]: ...

Related

Database Module — Pass DataAccess to DatabaseEventHandler for persistence
Storage Module — Archive event logs to cloud storage

---

db Database Module

PostgreSQL connection pool + Azure AD auth.

Configuration

AzureADConfig

Azure Active Directory authentication configuration.

@dataclass class AzureADConfig: tenant_id: str client_id: str client_secret: str # hidden username: str = "" @classmethod def from_dict(data: Dict[str, Any]) -> AzureADConfig: ... @classmethod def from_toml(toml_path: str, section: str = "azure_ad") -> AzureADConfig: ...

PostgreSQLConfig

PostgreSQL connection configuration (with Azure AD support).

class PostgreSQLConfig: provider: str = "postgresql" host: str = "localhost" port: int = 5432 database: str = "agenticstar_db" use_azure_ad: bool = False username: Optional[str]= None password: Optional[str]= None # hidden pool_min_size: int = 5 pool_max_size: int = 20 command_timeout: int = 60 pool_timeout: Optional[int]= 30 max_overflow: Optional[int]= 10 azure_ad: Optional[AzureADConfig]= None api_url: Optional[str]= None ssl_mode: str = "require" @classmethod def from_dict(data: Dict[str, Any]) -> PostgreSQLConfig: ... @classmethod def from_toml(toml_path: str, section: str = "database") -> PostgreSQLConfig: ... @classmethod def from_env(prefix: str = "DB_") -> PostgreSQLConfig: ...

TOML configuration example:

config.toml - PostgreSQLToml

[database]
provider = "postgresql"
host = "db.example.com"
port = 5432
database = "agenticstar_db"
use_azure_ad = true
pool_min_size = 5
pool_max_size = 20
command_timeout = 60

[azure_ad]
tenant_id = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
client_id = "yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy"
client_secret = "your-secret-here"
username = "dbuser@example.onmicrosoft.com"

Connection Management

PostgreSQLManager

Async PostgreSQL connection pool (with Azure AD token management).

class PostgreSQLManager: def __init__(config: PostgreSQLConfig): ... @property def config() -> PostgreSQLConfig: ... @property def pool() -> Optional[asyncpg.Pool]: ... async def __aenter__() -> PostgreSQLManager: ... async def __aexit__(...) -> None: ... def is_initialized() -> bool: ... async def initialize() -> None: ... async def close() -> None: ... async def execute_query( query: str, params: tuple = () ) -> Dict[str, Any]: ... async def fetch_one( query: str, params: tuple = () ) -> Optional[Dict[str, Any]]: ... async def fetch_all( query: str, params: tuple = () ) -> List[Dict[str, Any]]: ... async def execute( query: str, params: tuple = () ) -> str: ...

Returns

execute_query() -> Dict[str, Any]{"success": bool, "data": list (row list for SELECT / RETURNING) or {"result": str} (PostgreSQL status for other DML)}

fetch_one() -> Optional[Dict[str, Any]]Single row result dict, or None if not found

fetch_all() -> List[Dict[str, Any]]List of result rows (empty list if 0 rows)

execute() -> strPostgreSQL status string (e.g., "INSERT 0 1")

✓

Context manager supported: async with PostgreSQLManager(config) as mgr:

Database Connection and Query ExecutionPython

from agenticstar_platform.db import PostgreSQLConfig, PostgreSQLManager

config = PostgreSQLConfig.from_toml("config.toml")

async with PostgreSQLManager(config) as mgr:
    await mgr.initialize()

    # SELECT query
    result = await mgr.fetch_all(
        "SELECT * FROM conversations WHERE user_id = $1",
        ("user-123",)
    )

    # INSERT
    await mgr.execute(
        "INSERT INTO events (event_type, data) VALUES ($1, $2)",
        ("phase_start", '{}')
    )

Data Access Layer

DataAccess

Generic table operation wrapper with SQL injection prevention.

class DataAccess: def __init__( db: PostgreSQLManager| ApiPostgreSQLManager ): ... async def __aenter__() -> DataAccess: ... async def __aexit__(...) -> None: ... async def initialize() -> None: ... async def close() -> None: ... def is_initialized() -> bool: ... async def ensure_initialized() -> None: ... async def execute_query( query: str, params: tuple = () ) -> Dict[str, Any]: ... async def insert( table: str, data: Dict[str, Any], returning: Optional[List[str]]= None ) -> Dict[str, Any]: ... async def upsert( table: str, data: Dict[str, Any], conflict_columns: List[str], update_columns: Optional[List[str]]= None, returning: Optional[List[str]]= None ) -> Dict[str, Any]: ... async def select( table: str, columns: Optional[List[str]]= None, where: Optional[Dict[str, Any]]= None, order_by: Optional[str]= None, limit: Optional[int]= None ) -> Dict[str, Any]: ... async def select_one( table: str, columns: Optional[List[str]]= None, where: Optional[Dict[str, Any]]= None ) -> Optional[Dict[str, Any]]: ... async def update( table: str, data: Dict[str, Any], where: Dict[str, Any], returning: Optional[List[str]]= None ) -> Dict[str, Any]: ... async def delete( table: str, where: Dict[str, Any], returning: Optional[List[str]]= None ) -> Dict[str, Any]: ...

Returns

insert() / upsert() -> Dict[str, Any]{"success": bool, "data": list or {"result": str}}. A row list of column values when returning is specified, otherwise {"result": str}

select() -> Dict[str, Any]{"success": bool, "data": list}

update() / delete() -> Dict[str, Any]{"success": bool, "data": {"result": str}}. A row list when returning is specified

Key method usage examples:

# INSERT result = await da.insert("users", {"name": "Alice"}, returning=["id"]) # UPSERT (update on conflict) result = await da.upsert("users", {"email": "a@example.com", "name": "Alice"}, conflict_columns=["email"]) # SELECT with WHERE + ORDER + LIMIT result = await da.select("users", columns=["id", "name"], where={"active": True}, order_by="created_at DESC", limit=10)

ConfigAccess

Access layer for configuration information such as agents, tools, guardrails, and MCP.

class ConfigAccess: def __init__(db: DataAccess): ... async def get_agent_instructions(agent_type: str) -> Dict[str, Any]: ... async def get_all_agent_configs() -> Dict[str, Any]: ... async def get_agent_config(agent_type: str, requested_level: str = "default") -> Dict[str, Any]: ... async def get_tool_config(tool_name: str) -> Dict[str, Any]: ... async def get_all_tool_configs() -> Dict[str, Any]: ... async def get_banned_urls() -> Dict[str, Any]: ... async def get_guardrails_settings() -> Dict[str, Any]: ... async def get_system_settings() -> Dict[str, Any]: ... async def get_mcp_configurations(execution_mode: Optional[str]= None) -> Dict[str, Any]: ...

get_mcp_configurations() returns the active MCP server configurations as {"success": True, "data": [...]}. Each item in data has id / name / server_url / cache_tools_list / headers / only_pod / use_oauth / oauth_service / transport_type. id (the MCP server configuration ID) was added in v3.0.2+. name is a display name that administrators can set freely, so use id to identify or match a connection.

Extending Blocked URL granularity

get_banned_urls() returns the list of blocked URL patterns configured in the admin panel. The basic feature (ExtAuth Service) provides domain-level restriction, but by matching the retrieved patterns against URL strings inside your agent tools (e.g., using re.search), you can implement URL (path) level restriction as an extended capability. Settings are unified through the admin panel, so there is no duplicated operational configuration.

ExecutionAccess

Execution message retrieval layer.

class ExecutionAccess: def __init__(db: DataAccess): ... async def get_messages( execution_id: str ) -> Optional[List[Dict[str, Any]]]: ...

TelemetryAccess

NIST audit telemetry storage and retrieval layer.

class TelemetryAccess: def __init__(db: DataAccess): ... async def save_telemetry( telemetry_data: Dict[str, Any] ) -> Dict[str, Any]: ... async def list_telemetry( conversation_id: Optional[str]= None, service: Optional[str]= None, limit: int = 100, offset: int = 0 ) -> Dict[str, Any]: ... async def get_telemetry( telemetry_id: str ) -> Optional[Dict[str, Any]]: ...

PodRuntime

Pod lifecycle management (startup notification, shutdown notification, optional self scale-down).

class PodRuntime: def __init__( db: DataAccess, execution_id: str, pod_name: str, scale_down_callback: Optional[Callable[[str], Awaitable[None]]] = None ): ... async def start() -> Dict[str, Any]: ... async def final(status: str = "completed") -> Dict[str, Any]: ...

scale_down_callback is the self scale-down handler invoked at Pod termination (async def (execution_id: str) -> None). When it is not provided, final() only updates the status and does not perform self scale-down. Inject it when you want the Pod to terminate itself (e.g. delete this execution's SandboxClaim).

ApiPostgreSQLManager

PostgreSQL access via HTTP API proxy (for CLI mode). Has the same async interface as PostgreSQLManager.

v3.0.15+ — Query parameters are sent with a type marker for date / datetime values, so a plain string such as "2026-10-09" bound to a TEXT column is no longer turned into a date by the proxy. date / datetime values keep arriving as dates. Against a cli-api older than the matching one (Marketplace 3.1 and later) the proxy keeps guessing from the string shape as before.

class ApiPostgreSQLManager: def __init__(config: PostgreSQLConfig,*, token_provider: Callable[[], Optional[str]]): ... def is_initialized() -> bool: ... async def initialize() -> None: ... async def close() -> None: ... async def fetch_one(query: str, params: tuple = ()) -> Optional[Dict[str, Any]]: ... async def fetch_all(query: str, params: tuple = ()) -> List[Dict[str, Any]]: ... async def execute(query: str, params: tuple = ()) -> str: ... async def execute_query(query: str, params: tuple = ()) -> Dict[str, Any]: ...

Factory Function

def create_postgresql_manager( config: PostgreSQLConfig ) -> Union[PostgreSQLManager, ApiPostgreSQLManager]

Automatically selects and returns PostgreSQLManager (direct connection) or ApiPostgreSQLManager (API proxy) based on whether config.api_url is set.

Completion report for cycle executions

v3.0.14+ — In a cycle execution (execution_data.runtime_protocol = mp-chat/1, a conversation the platform starts on a schedule), your agent reports whether it achieved its goal. An execution without a report is recorded as "finished, completion not measured" (no outcome, runner_exited) even when it exits 0, and counts neither as a success nor as a failure. The platform never treats a clean process exit as completion. For an example, see the Quickstart.

async def report_completion( pool: asyncpg.Pool, # PostgreSQLManager.pool (direct connection pool) *, user_id: str, # USER_ID conversation_id: str, # CONVERSATION_ID message_id: str, # MESSAGE_ID execution_id: str, # EXECUTION_ID outcome: str, reason: Optional[str]= None, had_mutating_success: Optional[bool]= None, summary: Optional[str]= None, ) -> bool

Argument	Description
outcome	One of completed / incomplete / failed / timeout / stopped / denied / skipped. Anything else raises ValueError
reason	Reason code (outcome_reason_code): lowercase snake_case of at most 32 characters (^[a-z][a-z0-9_]{0,31}$; - and spaces become _, upper case becomes lower case). When omitted, or for any other value (a sentence, an exception string), the value of outcome is used and a warning is logged. Put free text in summary
had_mutating_success	Whether a write outside the agent succeeded (True / False; None if unknown). Kept in the report row (the platform currently uses only outcome and the reason code for the execution result)
summary	Optional note for people reading the record

• from agenticstar_platform import report_completion (also importable from agenticstar_platform.db). Requires the [db] extra (asyncpg)
• Returns True when the report was written and False when it was not
• Reporting is optional. Agents that do not report stay unmeasured, exactly as before
• Report once. If the execution already has a report, nothing is written, a warning is logged and False is returned (the first report is kept). Concurrent calls also write only one row
• A database failure never fails your agent. If the write fails (including a 30-second wait for a connection), a warning is logged, False is returned and the execution stays unmeasured. The only exception raised is for an outcome outside the 7 values. If only the commit acknowledgement is lost, False can be returned even though the report was saved
• The platform adopts the report only when it sees the container exit 0. When it closes the execution otherwise (a non-zero exit or a lost pod = failed with unknown effects, a cancellation = stopped/cancelled, the deadline passing while the agent is still running = timeout/execution_timeout), the report is not used. run_marketplace_agent exits 0 even when your agent raises, so a report already written is adopted in that case too
• The report is one row in execution_messages (content_data.kind = sdk.runner.completion, matched by the same CONVERSATION_ID / MESSAGE_ID as the started fence). No DDL is involved. It is written directly to the database because a cycle agent has no runtime API
• Calling it in a normal (chat) execution changes nothing (the report is not read). Admin dashboards count the row as one chunk of the execution, like the started fence

Error Handling

Database operation exception handling.

Database Error HandlingPython

from agenticstar_platform.db import PostgreSQLConfig, PostgreSQLManager, DataAccess

config = PostgreSQLConfig.from_toml("config.toml")

try:
    db_manager = PostgreSQLManager(config)
    da = DataAccess(db_manager)
    await da.initialize()

    result = await da.select("users", where={"id": "user-123"})
    if result["success"] and result["data"]:
        print(f"User: {result['data'][0]['username']}")
except Exception as e:
    print(f"Database error: {e}")
finally:
    await da.close()

Related

Events Module — Persist events to DB with DatabaseEventHandler
RAG Module — Convert data retrieved from DB into Embeddings for searchability

---

rag RAG Module

RAG (Retrieval-Augmented Generation): Embedding + Vector Search via Qdrant.

Enums

VectorStoreProvider

Vector store provider type for RAG.

QDRANT = "qdrant"

Exceptions

VectorStoreError (Exception) ├── VectorStoreConfigError │ └── QdrantConfigError └── VectorStoreConnectionError EmbeddingError (Exception) └── RateLimitExceededError

Data Classes

SearchResult

Single search result.

@dataclass class SearchResult: point_id: str payload: Dict[str, Any] score: float similarity: float = 0.0

UpsertResult

Upsert operation result.

@dataclass class UpsertResult: success: bool point_id: str = "" error: Optional[str]= None error_code: Optional[str]= None

SearchResponse

Search operation result.

@dataclass class SearchResponse: success: bool results: List[SearchResult]= field(default_factory=list) total_found: int = 0 query: str = "" error: Optional[str]= None error_code: Optional[str]= None

Embedding Configuration

EmbeddingConfig

Embedding configuration for Azure OpenAI / OpenAI-compatible endpoints.

provider = "openai" (v0.5.24+) targets OpenAI-compatible endpoints such as embed-v-4-0 on Azure AI inference (base_url must include /models; api_version is not used). Model strings with a LiteLLM-style prefix (azure/... / openai/...) derive the provider automatically, and the prefix is stripped from the deployment/model name.

class EmbeddingConfig: base_url: str api_key: str # hidden model: str = "text-embedding-ada-002" api_version: str = "2024-02-15-preview" max_cache_size: int = 10000 dimensions: int = 1536 provider: str = "azure" # "azure" | "openai" (v0.5.24+) max_retries: int = 5 base_delay: float = 1.0 max_delay: float = 60.0 @classmethod def from_dict(data: Dict[str, Any]) -> EmbeddingConfig: ... @classmethod def from_toml(toml_path: str, section: str = "rag.embedding") -> EmbeddingConfig: ...

EmbeddingGenerator

Embedding generation with caching and retry logic.

class EmbeddingGenerator: def __init__(config: EmbeddingConfig): ... async def generate(text: str) -> List[float]: ... async def batch_generate( texts: List[str], batch_size: int = 16 ) -> List[List[float]]: ... def clear_cache() -> None: ... def get_cache_stats() -> Dict[str, Any]: ...

Returns

generate() -> List[float]Vector (dimensions per config.dimensions, default 1536). Returns immediately if cached

batch_generate() -> List[List[float]]List of vectors in input text order. API calls in batch_size units (default 16)

get_cache_stats() -> Dict[str, Any]Cache statistics (cache_size, max_cache_size, cache_utilization, model_name)

# Single text Embedding vector = await generator.generate("Hello, world!") # -> [0.012, -0.034, ...] # Batch Embedding vectors = await generator.batch_generate(["text1", "text2", "text3"])

Vector Database

QdrantConfig

Qdrant database configuration.

class QdrantConfig: url: str collection_name: str vector_size: int = 1536 distance: str = "cosine" # cosine, euclid, dot on_disk_vectors: bool = False on_disk_payload: bool = True hnsw_m: int = 16 hnsw_ef_construct: int = 256 payload_indexes: List[PayloadIndexConfig] auth_token_provider: Optional[Callable[[], str]]= None prefer_grpc: bool = True check_compatibility: bool = True @classmethod def from_dict(data: Dict[str, Any]) -> QdrantConfig: ... @classmethod def from_toml(toml_path: str, section: str = "rag.qdrant") -> QdrantConfig: ...

PayloadIndexConfig

Payload field index configuration.

@dataclass class PayloadIndexConfig: field_name: str field_schema: str = "keyword" # keyword, integer, float, bool @classmethod def from_dict(data: Dict[str, Any]) -> PayloadIndexConfig: ...

TOML configuration example:

config.toml - RAG (Qdrant)Toml

[embedding]
provider = "openai"
model = "text-embedding-3-small"
api_key = "sk-..."
dimension = 1536

[qdrant]
host = "qdrant.example.com"
port = 6333
api_key = "your-qdrant-key"
use_ssl = true

VectorStoreClientBase

Abstract base class for vector-store clients (abc.ABC).

class VectorStoreClientBase(ABC): def __init__(): ... async def __aenter__() -> VectorStoreClientBase: ... async def __aexit__(...) -> None: ... def is_initialized() -> bool: ... async def ensure_initialized() -> None: ... @abstractmethod async def initialize() -> None: ... @abstractmethod async def upsert( point_id: str, text: str, payload: Dict[str, Any] ) -> Dict[str, Any]: ... @abstractmethod async def batch_upsert( items: List[Dict[str, Any]], batch_size: int = 256 ) -> Dict[str, Any]: ... @abstractmethod async def search( query_text: str, limit: int = 10, score_threshold: float = 0.4, filter_conditions: Optional[Dict[str, Any]]= None ) -> Dict[str, Any]: ... @abstractmethod async def delete( point_ids: List[str] ) -> Dict[str, Any]: ... @abstractmethod async def get_statistics() -> Dict[str, Any]: ... @abstractmethod async def close() -> None: ...

Returns

upsert() → Dict[str, Any]{"success": bool, "data": {"id": str}}

batch_upsert() → Dict[str, Any]{"success": bool, "data": {"total_upserted": int}}

search() → Dict[str, Any]{"success": bool, "data": {"query": str, "results": List[SearchResult], "total_found": int, "score_threshold": float}}

get_statistics() → Dict[str, Any]{"success": bool, "data": {"total_objects": int, "vectors_count": int, "content_type_breakdown": dict, "collection_name": str, "status": str}}

The base contract's upsert(point_id, text, payload) uses different parameter names than the concrete QdrantManager.upsert(id, content, metadata).

VectorStoreClientProtocol

Vector-store client protocol (typing.Protocol, @runtime_checkable).

@runtime_checkable class VectorStoreClientProtocol(Protocol): def is_initialized() -> bool: ... async def initialize() -> None: ... async def ensure_initialized() -> None: ... async def upsert( point_id: str, text: str, payload: Dict[str, Any] ) -> Dict[str, Any]: ... async def batch_upsert( items: List[Dict[str, Any]], batch_size: int = 256 ) -> Dict[str, Any]: ... async def search( query_text: str, limit: int = 10, score_threshold: float = 0.4, filter_conditions: Optional[Dict[str, Any]]= None ) -> Dict[str, Any]: ... async def delete( point_ids: List[str] ) -> Dict[str, Any]: ... async def get_statistics() -> Dict[str, Any]: ... async def close() -> None: ...

Returns

upsert() → Dict[str, Any]{"success": bool, "data": {"id": str}}

batch_upsert() → Dict[str, Any]{"success": bool, "data": {"total_upserted": int}}

search() → Dict[str, Any]{"success": bool, "data": {"query": str, "results": List[SearchResult], "total_found": int, "score_threshold": float}}

get_statistics() → Dict[str, Any]{"success": bool, "data": {"total_objects": int, "vectors_count": int, "content_type_breakdown": dict, "collection_name": str, "status": str}}

QdrantManager

Qdrant vector search client.

class QdrantManager: def __init__( config: QdrantConfig, embedding_generator: EmbeddingGenerator ): ... async def __aenter__() -> QdrantManager: ... async def __aexit__(...) -> None: ... def is_initialized() -> bool: ... async def initialize() -> None: ... async def ensure_initialized() -> None: ... async def upsert( id: str, content: str, metadata: Dict[str, Any] ) -> Dict[str, Any]: ... async def batch_upsert( items: List[Dict[str, Any]], batch_size: int = 256 ) -> Dict[str, Any]: ... async def search( query_text: str, limit: int = 10, score_threshold: float = 0.4, filter_conditions: Optional[Dict[str, Any]]= None, ef: Optional[int]= None ) -> Dict[str, Any]: ... async def delete( point_ids: List[str] ) -> Dict[str, Any]: ... async def get_statistics() -> Dict[str, Any]: ... async def collection_exists( collection_name: Optional[str]= None ) -> bool: ... async def create_payload_index( field_name: str, field_schema: str = "keyword" ) -> Dict[str, Any]: ... async def delete_collection( collection_name: Optional[str]= None ) -> Dict[str, Any]: ... async def close() -> None: ...

Returns

upsert() -> Dict[str, Any]{"success": bool, "data": {"id": str}}

batch_upsert() -> Dict[str, Any]{"success": bool, "data": {"total_upserted": int}}

search() -> Dict[str, Any]{"success": bool, "data": {"query": str, "results": List[SearchResult], "total_found": int, "score_threshold": float}}

get_statistics() -> Dict[str, Any]{"success": bool, "data": {"total_objects": int, "vectors_count": int, "content_type_breakdown": dict, "collection_name": str, "status": str}}

collection_exists() -> boolTrue if the collection exists. Falls back to `config.collection_name` when `collection_name` is omitted

create_payload_index() -> Dict[str, Any]Create an index on a payload field. `field_schema` accepts "keyword" / "integer" / "float" / "bool" etc.

delete_collection() -> Dict[str, Any]Delete the collection. Falls back to `config.collection_name` when `collection_name` is omitted

✓

Context manager supported: async with QdrantManager(config, generator) as qdrant:

RAG: Embedding + Qdrant SearchPython

from agenticstar_platform.rag import (
    EmbeddingConfig, EmbeddingGenerator,
    QdrantConfig, QdrantManager
)

emb_config = EmbeddingConfig.from_toml("config.toml")
qd_config = QdrantConfig.from_toml("config.toml")

generator = EmbeddingGenerator(emb_config)

async with QdrantManager(qd_config, generator) as qdrant:
    # Add a document
    await qdrant.upsert(
        id="doc-001",
        content="AGENTIC STAR is an AI agent development platform",
        metadata={"source": "docs", "category": "overview"}
    )

    # Semantic search
    results = await qdrant.search(
        query_text="What is an AI platform?",
        limit=5,
        score_threshold=0.5
    )
    for r in results["data"]:
        print(f"Score: {r['score']}, Text: {r['payload']['text']}")

Related

Database Module — Feed data retrieved from DB into RAG
Storage Module — Retrieve document files from storage and convert to Embeddings
Security Module — Verify content safety before Embedding

---

storage Storage Module

Cloud storage integration (Azure Blob Storage / AWS S3 / Google Cloud Storage).

Enums

StorageProvider

Cloud storage provider type.

AZURE_BLOB = "azure_blob"

AWS_S3 = "aws_s3"

GCS = "gcs"

Exceptions

StorageError (Exception) - Base exception ├── StorageConfigError - Configuration error ├── StorageConnectionError - Connection error └── StorageOperationError - Operation error with error_code attribute

Result Types

UploadResult

File upload result.

@dataclass class UploadResult: success: bool # Upload succeeded object_name: str = "" # Object name object_url: str = "" # Object URL file_size: int = 0 # File size content_type: str = "" # Content type error: Optional[str]= None # Error message error_code: Optional[str]= None # Error code metadata: Dict[str, Any]= field(default_factory=dict) # Metadata

DownloadResult

File download result.

@dataclass class DownloadResult: success: bool # Download succeeded object_name: str = "" # Object name local_path: str = "" # Local file path file_size: int = 0 # File size error: Optional[str]= None # Error message error_code: Optional[str]= None # Error code

ObjectInfo

Cloud object metadata.

@dataclass class ObjectInfo: name: str # Object name size: int # Object size last_modified: Optional[str]= None # Last modified timestamp content_type: Optional[str]= None # Content type metadata: Dict[str, Any]= field(default_factory=dict) # Metadata

ListResult

List operation result.

@dataclass class ListResult: success: bool # Operation succeeded objects: List[ObjectInfo]= field(default_factory=list) # Objects count: int = 0 # Object count prefix: str = "" # Search prefix error: Optional[str]= None # Error message error_code: Optional[str]= None # Error code

Configuration

StorageConfig

Common storage configuration (base class).

class StorageConfig: provider: StorageProvider# Provider type bucket_name: str # Bucket/container name enabled: bool = True # Enabled flag max_file_size: int = 5*1024*1024*1024 # Max file size (5GB) auto_create_bucket: bool = False # Auto-create bucket prefix: str = "" # Object name prefix custom_domain: Optional[str]= None # Custom domain connection_timeout: int = 30 # Connection timeout (seconds) read_timeout: int = 300 # Read timeout (seconds)

AzureBlobConfig

Azure Blob Storage configuration.

class AzureBlobConfig: bucket_name: str # Container name connection_string: str # hidden enabled: bool = True max_file_size: int = 5*1024*1024*1024 auto_create_bucket: bool = False prefix: str = "" custom_domain: Optional[str]= None connection_timeout: int = 30 read_timeout: int = 300 max_block_size: int = 8*1024*1024 # Max block size (8MB) max_single_put_size: int = 256*1024*1024 # Max single upload (256MB) @classmethod def from_dict(data: Dict[str, Any]) -> AzureBlobConfig: ... @property def provider(self) -> StorageProvider: return StorageProvider.AZURE_BLOB

S3Config

AWS S3 configuration.

class S3Config: bucket_name: str # Bucket name aws_access_key_id: str # hidden aws_secret_access_key: str # hidden region_name: str = "us-east-1" # AWS region enabled: bool = True max_file_size: int = 5*1024*1024*1024 auto_create_bucket: bool = False prefix: str = "" custom_domain: Optional[str]= None connection_timeout: int = 30 read_timeout: int = 300 endpoint_url: Optional[str]= None # S3-compatible endpoint (MinIO, etc.) @classmethod def from_dict(data: Dict[str, Any]) -> S3Config: ... @property def provider(self) -> StorageProvider: return StorageProvider.AWS_S3

GCSConfig

Google Cloud Storage configuration.

class GCSConfig: bucket_name: str # Bucket name project_id: str # GCP project ID enabled: bool = True max_file_size: int = 5*1024*1024*1024 auto_create_bucket: bool = False prefix: str = "" custom_domain: Optional[str]= None connection_timeout: int = 30 read_timeout: int = 300 credentials_path: Optional[str]= None # hidden - Path to service account JSON credentials_json: Optional[str]= None # hidden - Service account JSON string credentials_base64: Optional[str]= None # hidden - Base64-encoded service account JSON @classmethod def from_dict(data: Dict[str, Any]) -> GCSConfig: ... @property def provider(self) -> StorageProvider: return StorageProvider.GCS

Storage Clients

AzureBlobStorageClient

Azure Blob Storage client. Supports async context manager (async with).

class AzureBlobStorageClient: def __init__(config: AzureBlobConfig): ... @property def provider(self) -> StorageProvider: return StorageProvider.AZURE_BLOB @property def container_name(self) -> str: ... async def ensure_bucket_exists() -> bool: ... async def upload_file( file_path: str, object_name: Optional[str]= None, prefix: Optional[str]= None, metadata: Optional[Dict[str, Any]]= None ) -> UploadResult: ... async def download_file( object_name: str, download_path: str ) -> DownloadResult: ... async def list_objects( prefix: str = "", max_results: Optional[int]= None ) -> ListResult: ... async def delete_object( object_name: str ) -> bool: ... async def object_exists( object_name: str ) -> bool: ... async def close() -> None: ...

Returns

upload_file() -> UploadResultContains success, object_url, file_size, content_type

download_file() -> DownloadResultContains success, local_path, file_size

list_objects() -> ListResultContains success, objects: List[ObjectInfo], count

delete_object() -> boolTrue if the object was deleted

S3StorageClient

AWS S3 storage client. Supports async context manager (async with).

class S3StorageClient: def __init__(config: S3Config): ... @property def provider(self) -> StorageProvider: return StorageProvider.AWS_S3 @property def bucket_name(self) -> str: ... async def ensure_bucket_exists() -> bool: ... async def upload_file( file_path: str, object_name: Optional[str]= None, prefix: Optional[str]= None, metadata: Optional[Dict[str, Any]]= None ) -> UploadResult: ... async def download_file( object_name: str, download_path: str ) -> DownloadResult: ... async def list_objects( prefix: str = "", max_results: Optional[int]= None ) -> ListResult: ... async def delete_object( object_name: str ) -> bool: ... async def object_exists( object_name: str ) -> bool: ... async def close() -> None: ...

GCSStorageClient

Google Cloud Storage client. Supports async context manager (async with).

class GCSStorageClient: def __init__(config: GCSConfig): ... @property def provider(self) -> StorageProvider: return StorageProvider.GCS @property def bucket_name(self) -> str: ... async def ensure_bucket_exists() -> bool: ... async def upload_file( file_path: str, object_name: Optional[str]= None, prefix: Optional[str]= None, metadata: Optional[Dict[str, Any]]= None ) -> UploadResult: ... async def download_file( object_name: str, download_path: str ) -> DownloadResult: ... async def list_objects( prefix: str = "", max_results: Optional[int]= None ) -> ListResult: ... async def delete_object( object_name: str ) -> bool: ... async def object_exists( object_name: str ) -> bool: ... async def close() -> None: ...

S3 Storage Usage ExamplePython

from agenticstar_platform.storage import (
    S3Config, S3StorageClient
)

config = S3Config.from_dict({
    "bucket_name": "my-bucket",
    "aws_access_key_id": "AKIA...",
    "aws_secret_access_key": "secret...",
    "region_name": "ap-northeast-1"
})

# Context manager with async with
async with S3StorageClient(config) as storage:
    # File upload
    upload_result = await storage.upload_file(
        file_path="/tmp/report.pdf",
        object_name="report.pdf",
        prefix="documents/",
        metadata={"source": "documents"}
    )
    if upload_result.success:
        print(f"URL: {upload_result.object_url}")

    # List objects
    list_result = await storage.list_objects(prefix="documents/")
    for obj in list_result.objects:
        print(f"{obj.name} ({obj.size} bytes)")

    # Download
    download = await storage.download_file(
        object_name="documents/report.pdf",
        download_path="/tmp/downloaded.pdf"
    )

Error HandlingPython

from agenticstar_platform.storage import (
    StorageError, StorageConfigError,
    StorageConnectionError, StorageOperationError
)

try:
    result = await storage.upload_file("/tmp/large-file.zip")
except StorageConnectionError as e:
    print(f"Connection failed: {e}")
except StorageOperationError as e:
    print(f"Operation failed: {e}, code: {e.error_code}")
except StorageError as e:
    print(f"Storage error: {e}")

Related

RAG Module — Retrieve documents from storage and feed into Embedding
Security Module — Content moderation before upload

---

auth Auth Module

AGENTIC STAR authentication service integration. Manages users, tokens, and device information.

Exceptions

AuthError (Exception) ├── AuthConfigError └── AuthAPIError ├── AuthUnauthorizedError (401) ├── AuthNotFoundError (404) └── AuthRateLimitError (429)

Configuration

AgenticStarAuthConfig

AGENTIC STAR Auth API client connection configuration.

class AgenticStarAuthConfig: base_url: str api_key: str # hidden auth_type: str = "bearer" timeout: float = 30.0 token_provider: Optional[Callable[[], Optional[str]]]= None @classmethod def create( base_url: str, api_key: str, auth_type: str = "bearer", timeout: float = 30.0, token_provider: Optional[Callable[[], Optional[str]]]= None ) -> AgenticStarAuthConfig: ... @classmethod def from_config(config_path: Optional[Path]= None) -> AgenticStarAuthConfig: ...

Data Models

ApiUser

Authenticated user information (Pydantic BaseModel).

class ApiUser(BaseModel): id: str username: Optional[str]= None email: Optional[str]= None first_name: Optional[str]= None last_name: Optional[str]= None display_name: Optional[str]= None email_verified: bool = False enabled: bool = True created_timestamp: Optional[int]= None attributes: Optional[Dict[str, Any]]= None organization: Optional[str]= None organization_label: Optional[str]= None job: Optional[str]= None job_label: Optional[str]= None bio: Optional[str]= None last_login_at: Optional[str]= None

DeviceInfo

Device information.

class DeviceInfo(BaseModel): id: Optional[str]= None device_code: str device_name: Optional[str]= None device_type: Optional[str]= None device_info: Optional[Dict[str, Any]]= None is_active: Optional[bool]= None created_at: Optional[str]= None last_used_at: Optional[str]= None

MCPTokenInfo

MCP token information. expires_at of None means "no expiry" (for personal-token (PAT) providers, the token supply API returns no expiry. v0.5.34+).

class MCPTokenInfo(BaseModel): access_token: str token_type: str expires_at: Optional[datetime]= None scopes: List[str]

UserPagination

Pagination information (the pagination field of get_users()).

class UserPagination(BaseModel): page: int limit: int total: int total_pages: int has_next: bool has_previous: bool

LoginHistoryEntry

A login-history entry (an element of get_user()'s login_history).

class LoginHistoryEntry(BaseModel): timestamp: int ip_address: str user_agent: str platform: str location: Optional[Dict[str, Optional[str]]]= None success: bool

MCPTokenError

An MCP token retrieval error (a value of get_mcp_tokens()'s errors). code is one of CONNECTION_NOT_FOUND / TOKEN_EXPIRED / REFRESH_FAILED, etc.

class MCPTokenError(BaseModel): code: str message: str

OAuthProviderName

A type alias for the OAuth provider name. Its value comes from the DB oauth_providers.service; it is the identifier passed to get_mcp_tokens(providers=[...]) and the dict key of tokens / errors (e.g. github, slack).

Python

OAuthProviderName = str

Client

AgenticStarAuthClient

HTTP client for the AGENTIC STAR authentication API.

class AgenticStarAuthClient: def __init__(config: AgenticStarAuthConfig): ... async def __aenter__() -> AgenticStarAuthClient: ... async def __aexit__(...) -> None: ... async def get_users( page: int = 1, limit: int = 20, search: Optional[str]= None, is_approved: Optional[str]= None, job: Optional[str]= None, organization: Optional[str]= None, sort_by: Optional[str]= None, order: Optional[str]= None, include_last_login: bool = False ) -> GetUsersResult: ... async def get_user( user_id: str ) -> GetUserResult: ... async def get_mcp_tokens( user_id: str, providers: Optional[List[str]]= None, rejected_token_sha256: Optional[Dict[str, str]]= None # v3.0.10+ ) -> GetMCPTokensResult: ... async def close() -> None: ...

Returns

get_users() -> GetUsersResultsuccess, users: List[ApiUser], pagination: UserPagination, error, error_code

get_user() -> GetUserResultsuccess, user: ApiUser, devices: List[DeviceInfo], login_history, is_deleted, deleted_at, deleted_by, deletion_reason, error, error_code

get_mcp_tokens() -> GetMCPTokensResultsuccess, tokens: Dict[str, MCPTokenInfo], errors, error, error_code

get_user() URL-encodes user_id as a single path segment (values containing / or ? never reach a different API path; v3.0.5+). An empty string and . / .. (v3.0.5+) return success=False with error_code="INVALID_PARAMETER" without sending a request.

rejected_token_sha256 of get_mcp_tokens() (v3.0.10+) passes, per provider, the SHA-256 of an access token that the MCP server rejected with 401 (hash the token string as UTF-8; lowercase hex, 64 characters). If the stored token is still that token, it is refreshed even before its expiry; if it has already been replaced, the newer token is returned. Personal access tokens (PAT), connections that cannot be refreshed, and tokens issued less than a minute ago are not refreshed (the stored token is returned). Do not pass the token itself. An auth service that does not support the field ignores it.

# Get user list (with pagination) result = await auth_client.get_users(page=1, limit=20) for user in result.users: print(f"{user.username} ({user.email})")

User Information and MCP TokensPython

from agenticstar_platform.auth import (
    AgenticStarAuthConfig, AgenticStarAuthClient
)

config = AgenticStarAuthConfig.from_config("config.toml")

async with AgenticStarAuthClient(config) as auth_client:
    # Get user information
    result = await auth_client.get_user("user-123")
    if result.success:
        print(f"User: {result.user.username}")

    # Get MCP tokens
    tokens = await auth_client.get_mcp_tokens(
        user_id="user-123",
        providers=["slack", "gitlab"],
    )
    for provider, token in tokens.tokens.items():
        print(f"{provider}: expires={token.expires_at}")

Related

Events Module — Supply auth tokens to WebhookEventHandler token_provider
Memory Module — Manage semantic memory for authenticated users

---

memory Memory Module

Semantic memory (Mem0 + Qdrant) integration. Recall user knowledge, preferences, and facts via vector search.

Configuration

SemanticMemoryConfig

Semantic memory configuration for Mem0 + Qdrant.

@dataclass class LLMProviderConfig: model: str # "azure_openai/gpt-4.1" api_key: str # hidden base_url: Optional[str]= None api_version: Optional[str]= None aws_access_key_id: Optional[str]= None # For Bedrock aws_secret_access_key: Optional[str]= None # For Bedrock aws_region_name: Optional[str]= None # For Bedrock @classmethod def from_dict(data: Dict[str, Any]) -> LLMProviderConfig: ... @classmethod def from_toml(toml_path: str, section: str) -> LLMProviderConfig: ... @classmethod def from_env(prefix: str = "LLM_") -> LLMProviderConfig: ...

@dataclass class SemanticMemoryConfig: llm_config: LLMProviderConfig embedder_config: LLMProviderConfig vector_store: Optional[Dict[str, Any]]= None rerank: Optional[Dict[str, Any]]= None custom_fact_extraction_prompt: Optional[str]= None custom_update_memory_prompt: Optional[str]= None @classmethod def from_dict(data: Dict[str, Any]) -> SemanticMemoryConfig: ... @classmethod def from_toml(toml_path: str, section: str = "memory") -> SemanticMemoryConfig: ...

QdrantVectorStoreConfig

Helper dataclass for assembling a Qdrant Vector Store configuration.

@dataclass class QdrantVectorStoreConfig: url: str collection_name: str embedding_dims: int = 1536 def to_dict() -> Dict[str, Any]: ... @classmethod def from_dict(data: Dict[str, Any]) -> QdrantVectorStoreConfig: ...

SemanticMemoryClient

Semantic memory client for add, search, and delete operations (synchronous API).

class SemanticMemoryClient: def __init__(config: SemanticMemoryConfig): ... @property def enabled() -> bool: ... # Synchronous methods (no await needed) def add( messages: List[Dict[str, str]], user_id: str, metadata: Optional[Dict[str, Any]]= None ) -> Dict[str, Any]: ... def search( query: str, user_id: str, limit: int = 10 ) -> Dict[str, Any]: ... def get_all( user_id: str ) -> Dict[str, Any]: ... def delete( memory_id: str ) -> Dict[str, Any]: ... def delete_all( user_id: str ) -> Dict[str, Any]: ... # Asynchronous method async def cleanup() -> None: ...

Returns

add() -> Dict[str, Any]Mem0 result dict. messages is OpenAI format [{"role": "user", "content": "..."}]

search() -> Dict[str, Any]{"results": [...]}. Each element contains memory, score, etc.

get_all() -> Dict[str, Any]Returns all memories for the user

delete() / delete_all() -> Dict[str, Any]Deletion result

Exceptions

SemanticMemoryError (Exception) └── SemanticMemoryConfigError

Utility Functions

def normalize_provider(provider: str) -> str

Normalize provider name (alias support, e.g., "Azure_OpenAI" -> "azure").

def parse_model_string(model: str) -> tuple[str, str]

Split model string into provider and model name (e.g., "azure_openai/gpt-4.1" -> ("azure", "gpt-4.1")).

def get_api_key(config: LLMProviderConfig) -> str

Get API key from LLMProviderConfig (SecretStr compatible).

def convert_llm_to_mem0(llm_config: LLMProviderConfig) -> Dict[str, Any]

Convert LLMProviderConfig to Mem0 format LLM configuration dict.

def convert_embedder_to_mem0(embedder_config: LLMProviderConfig) -> Dict[str, Any]

Convert LLMProviderConfig to Mem0 format Embedder configuration dict. For the openai provider (openai/... models), a configured base_url is passed through as openai_base_url (v0.5.24+; for OpenAI-compatible endpoints such as embed-v-4-0 — without it the client connects to the official api.openai.com). Versions up to v0.5.35 had a bug where this passthrough was not applied on the SemanticMemoryClient internal conversion path; fixed in v0.5.36.

Semantic Memory Usage ExamplePython

from agenticstar_platform.memory import (
    SemanticMemoryConfig, SemanticMemoryClient
)

config = SemanticMemoryConfig.from_toml("config.toml")
memory = SemanticMemoryClient(config)

# Add memory (synchronous method)
memory.add(
    messages=[
        {"role": "user", "content": "I'm proficient in Python"},
        {"role": "assistant", "content": "Understood"},
    ],
    user_id="user-123",
)

# Search memory (synchronous method)
result = memory.search(query="What are the user's technical skills?", user_id="user-123")
for r in result.get("results", []):
    print(f"{r['memory']}")

# Cleanup (asynchronous)
await memory.cleanup()

Related

Auth Module — Use authenticated user ID as memory user_id
RAG Module — Convert memory contents to Embeddings for enhanced similarity search

---

security Security Module

Content Moderation (Azure / AWS / GCP) + PII Detection. On AWS, PII detection can use either Bedrock Guardrails sensitiveInformationPolicy (multi-language, including Japanese) or Amazon Comprehend (en/es), selectable via AWSSecurityConfig.pii_service.

Enums

SecurityProvider

Security provider type.

AZURE = "azure"

AWS = "aws"

GCP = "gcp"

ContentCategory

Content moderation categories.

HATE = "hate"

SEXUAL = "sexual"

SELF_HARM = "self_harm"

VIOLENCE = "violence"

PROFANITY = "profanity"

INSULT = "insult"

THREAT = "threat"

PIICategory

PII (Personally Identifiable Information) categories.

PERSON_NAME = "person_name"

EMAIL = "email"

PHONE_NUMBER = "phone_number"

ADDRESS = "address"

AGE = "age"

DATE_OF_BIRTH = "date_of_birth"

NATIONAL_ID = "national_id"

PASSPORT_NUMBER = "passport_number"

DRIVERS_LICENSE = "drivers_license"

SOCIAL_SECURITY = "social_security"

TAX_ID = "tax_id"

CREDIT_CARD = "credit_card"

BANK_ACCOUNT = "bank_account"

SWIFT_CODE = "swift_code"

IBAN = "iban"

MEDICAL_RECORD = "medical_record"

HEALTH_INSURANCE = "health_insurance"

IP_ADDRESS = "ip_address"

MAC_ADDRESS = "mac_address"

URL = "url"

PASSWORD = "password"

API_KEY = "api_key"

AWS_ACCESS_KEY = "aws_access_key"

CONNECTION_STRING = "connection_string"

OTHER = "other"

Exceptions

SecurityError (Exception) ├── SecurityConfigError └── SecurityAPIError (status_code, error_code)

Result Types

ContentModerationResult

Content moderation result. categories values are severity (0-6).

@dataclass class ContentModerationResult: blocked: bool # True if above threshold categories: Dict[ContentCategory, int]= {} # severity 0-6 threshold: int = 2 # Applied threshold raw_response: Dict[str, Any]= {} error: Optional[str]= None error_code: Optional[str]= None

PromptShieldResult

Prompt injection detection result.

@dataclass class PromptShieldResult: attack_detected: bool attack_type: Optional[str]= None confidence: float = 0.0 raw_response: Dict[str, Any]= {} error: Optional[str]= None error_code: Optional[str]= None user_prompt_attack: Optional[bool]= None # v0.5.40+ direct (userPrompt) attack breakdown documents_attack: Optional[bool]= None # v0.5.40+ indirect (documents) attack breakdown

user_prompt_attack / documents_attack break the attack down by vector (v0.5.40+). None means "not captured" (older versions / not evaluated) and is distinct from False (evaluated, no attack). When both are known, attack_detected == (user_prompt_attack or documents_attack) holds.

PIIEntity

Detected PII entity.

@dataclass class PIIEntity: category: PIICategory text: str # Detected text offset: int # Start position in text length: int # Length of detected text confidence: float provider_category: str = "" # Provider-specific category name

PIIDetectionResult

PII detection result.

@dataclass class PIIDetectionResult: success: bool masked_text: str = "" entities: List[PIIEntity]= [] categories_detected: List[PIICategory]= [] raw_response: Dict[str, Any]= {} error: Optional[str]= None error_code: Optional[str]= None

SecurityCheckResult

Integrated security check result.

@dataclass class SecurityCheckResult: allowed: bool # Allowed / Blocked violations: List[str]= [] # List of violation reasons content_moderation: Optional[ContentModerationResult]= None prompt_shield: Optional[PromptShieldResult]= None pii_detection: Optional[PIIDetectionResult]= None

SanitizationResult

Sanitization result of ContentSafetyValidator.validate_and_sanitize(). marked_content is an empty string when the content is judged unsafe.

@dataclass class SanitizationResult: is_safe: bool marked_content: str = "" metadata: Dict[str, Any]= {}

Configuration

AzureSecurityConfig

Azure Content Safety + Language Service connection configuration.

class AzureSecurityConfig: content_safety_endpoint: str content_safety_api_key: str # hidden language_endpoint: Optional[str]= None # For PII detection language_api_key: Optional[str]= None # hidden enabled: bool = True prompt_shield_enabled: bool = True # Enable Prompt Shield moderation_enabled: bool = True # Enable content moderation moderation_threshold: int = 2 # severity 0-6 pii_enabled: bool = True pii_confidence_threshold: float = 0.7 pii_mask_string: str = "***" # PII mask string timeout: float = 30.0

AWSSecurityConfig

Security configuration for AWS Bedrock Guardrails / Comprehend.

class AWSSecurityConfig: aws_access_key_id: str = "" # hidden aws_secret_access_key: str = "" # hidden region_name: str = "us-east-1" guardrail_id: str = "" guardrail_version: str = "DRAFT" pii_service: str = "bedrock_guardrails" # PII backend: "bedrock_guardrails" / "comprehend" enabled: bool = False prompt_shield_enabled: bool = True # Enable Prompt Shield moderation_enabled: bool = True # Enable content moderation moderation_threshold: int = 2 pii_enabled: bool = True pii_confidence_threshold: float = 0.7 pii_mask_string: str = "***" # PII mask string timeout: float = 30.0

GCPSecurityConfig

Security configuration for GCP Model Armor + DLP.

class GCPSecurityConfig: project_id: str credentials_path: Optional[str]= None # hidden # Content Safety backend: "model_armor" (default) | "vertex_ai_safety" (all regions) content_safety_backend: str = "model_armor" model_armor_template: str = "" model_armor_region: str = "" # Vertex AI Safety Filters + Gemini judge (when content_safety_backend="vertex_ai_safety") vertex_ai_location: str = "global" vertex_harm_thresholds: Dict[str, str]= {} vertex_judge_model: str = "gemini-2.5-flash" dlp_location: str = "global" enabled: bool = False prompt_shield_enabled: bool = True # Enable Prompt Shield moderation_enabled: bool = True # Enable content moderation moderation_threshold: int = 2 pii_enabled: bool = True pii_confidence_threshold: float = 0.7 pii_mask_string: str = "***" # PII mask string timeout: float = 30.0

GuardrailsConfig

Guardrails configuration for ContentSafetyValidator (values intended to be set by marketplace customers via the admin UI). It can be created from a DB guardrails_settings dict via from_dict(). moderation_threshold is severity 0-6 (0 = disabled, 2 = recommended, 6 = extreme only).

@dataclass class GuardrailsConfig: prompt_shield_enabled: bool = True moderation_threshold: int = 2 fail_on_error: bool = True mark_content: bool = True @classmethod def from_dict(data: Dict[str, Any]) -> GuardrailsConfig: ...

Clients

SecurityClientBase

Abstract base class for all security clients (Azure / AWS / GCP). enabled reflects the configured enabled state, and provider is an abstract property implemented by subclasses (returns the provider type).

class SecurityClientBase: @property def enabled() -> bool: ... @property def provider() -> SecurityProvider: ...

AzureSecurityClient

Content moderation (Azure Content Safety) and PII detection (Language Service) client.

class AzureSecurityClient(SecurityClientBase): def __init__(config: AzureSecurityConfig): ... async def __aenter__() -> AzureSecurityClient: ... async def __aexit__(...) -> None: ... @property def enabled() -> bool: ... @property def provider(self) -> SecurityProvider: return SecurityProvider.AZURE async def check_content_moderation( text: str, threshold: Optional[int]= None ) -> ContentModerationResult: ... async def check_prompt_shield( user_prompt: str, documents: Optional[List[str]]= None ) -> PromptShieldResult: ... async def detect_pii( text: str, mask: bool = True, language: str = "ja", confidence_threshold: Optional[float]= None ) -> PIIDetectionResult: ... async def detect_pii_batch( # v0.5.23+ texts: List[str], mask: bool = True, language: str = "ja", confidence_threshold: Optional[float]= None ) -> List[PIIDetectionResult]: ... async def check_security( text: str, check_moderation: bool = True, check_prompt_shield: bool = True, check_pii: bool = False, fail_on_error: bool = True ) -> SecurityCheckResult: ... async def close() -> None: ...

Returns

check_content_moderation() -> ContentModerationResultblocked (block decision), categories (severity 0-6)

check_prompt_shield() -> PromptShieldResultattack_detected, attack_type, confidence

detect_pii() -> PIIDetectionResultsuccess, entities (list of PIIEntity), masked_text

detect_pii_batch() -> List[PIIDetectionResult]Same order/length as input. Azure packs 5 documents/request (up to 5x fewer calls). 429s are retried honoring Retry-After; persistent throttling fails texts closed with error_code="RATE_LIMITED" (v0.5.23+)

check_security() -> SecurityCheckResultallowed, violations, individual detection results

# Using PII mask results result = await security.detect_pii("Email: user@example.com", mask=True) print(result.masked_text) # -> "Email: ***" print(result.entities[0].category) # -> PIICategory.EMAIL

AWSSecurityClient / GCPSecurityClient

Same interface as Azure (inherits SecurityClientBase). Implements check_content_moderation(), check_prompt_shield(), detect_pii(), check_security(), close(). detect_pii_batch() is inherited from the base class as a sequential default (provider-side batching is Azure-only, v0.5.23+).

Content Moderation + PII DetectionPython

from agenticstar_platform.security import (
    AzureSecurityConfig, AzureSecurityClient
)

config = AzureSecurityConfig(
    content_safety_endpoint="https://your-cs.cognitiveservices.azure.com/",
    content_safety_api_key="your-key",
)

async with AzureSecurityClient(config) as security:
    # Content moderation
    moderation = await security.check_content_moderation(
        text="This is a sample message",
        threshold=2,
    )

    if moderation.blocked:
        print(f"Content blocked: {moderation.categories}")

    # Integrated check
    result = await security.check_security(
        text="Contact me at user@example.com or 123-456-7890",
        check_pii=True,
    )

    if not result.allowed:
        print(f"Violations: {result.violations}")
    if result.pii_detection and result.pii_detection.entities:
        print(f"PII found: {result.pii_detection.masked_text}")

ContentSafetyValidator

Wrapper that validates externally retrieved content (Web / MCP, etc.) using a SecurityClientProtocol implementation (e.g. AzureSecurityClient) and sanitizes/marks it based on GuardrailsConfig. validate_and_sanitize() handles a single piece of content, while validate_multiple() processes multiple pieces in a batch. mark_content() / unmark_content() are static methods that add/remove the marker ([EXT:...]...[/EXT]).

class ContentSafetyValidator: def __init__( security_client: SecurityClientProtocol, guardrails: Optional[GuardrailsConfig]= None ): ... @property def guardrails() -> GuardrailsConfig: ... def update_guardrails(guardrails: GuardrailsConfig) -> None: ... async def validate_and_sanitize( content: str, source_type: str, source_url: Optional[str]= None, user_prompt: Optional[str]= None ) -> SanitizationResult: ... async def validate_multiple( contents: List[Tuple[str, str, Optional[str]]], user_prompt: Optional[str]= None ) -> Tuple[bool, List[str], List[Dict[str, Any]]]: ... @staticmethod def mark_content( content: str, source_type: str, source_url: Optional[str]= None ) -> str: ... @staticmethod def unmark_content(marked_content: str) -> Tuple[str, Optional[str], Optional[str]]: ...

Returns

validate_and_sanitize() -> SanitizationResultis_safe (safety decision), marked_content (marked content), metadata

validate_multiple() -> Tuple[bool, List[str], List[Dict[str, Any]]]overall safety decision, list of marked content, list of metadata

mark_content() -> strmarked content with source information added

unmark_content() -> Tuple[str, Optional[str], Optional[str]]original content, source_type, source_url

Factory Functions

Module-level functions that auto-detect the provider from a config JSON and build the appropriate SecurityClient.

def detect_provider(config: Dict[str, Any]) -> SecurityProvider

Auto-detects the security provider (Azure / AWS / GCP) from a config JSON. Raises SecurityConfigError if it cannot be determined.

def create_security_client( config: Dict[str, Any], pii_config: Optional[Dict[str, Any]]= None ) -> SecurityClientBase

Builds the matching SecurityClient from a config JSON. Passing pii_config merges the content_safety and pii_language settings (Azure).

def create_content_safety_client(config: Dict[str, Any]) -> SecurityClientBase

Shortcut for Content Moderation / Prompt Shield. Delegates to create_security_client(config).

def create_pii_client(config: Dict[str, Any]) -> SecurityClientBase

Shortcut for PII detection / masking. For Azure, wires the config as the Language Service endpoint.

Related

RAG Module — Content validation pipeline before Embedding ingestion
Storage Module — PII screening before file upload
Events Module — Deliver moderation results as events

---

common Common Module

Common utilities. SQL injection prevention, secret masking, async context manager mixin.

SecretMasker

Utility for masking secret values (passwords, tokens, etc.) based on key names.

class SecretMasker: def __init__( secret_keys: Optional[Set[str]]= None, mask_string: str = "***", case_sensitive: bool = False, ): ... def mask_value(key: str, value: Any) -> Any: ... def mask_dict(data: Dict[str, Any]) -> Dict[str, Any]: ...

Returns

mask_value() → AnyReturns mask_string if key is a secret key, otherwise the original value

mask_dict() → Dict[str, Any]Recursively walks the dict and returns a new dict with secret values masked

def mask_secret(value: str, visible_chars: int = 0, mask_string: str = "***") -> str

Shortcut function for masking a single string.

Secret MaskingPython

from agenticstar_platform.common import SecretMasker, mask_secret

masker = SecretMasker()
safe = masker.mask_dict({
    "username": "alice",
    "password": "p@ssw0rd",
    "api_key": "sk-12345",
})
# -> {"username": "alice", "password": "***", "api_key": "***"}

mask_secret("my-secret-api-key-12345")                   # -> "***"
mask_secret("my-secret-api-key-12345", visible_chars=4)  # -> "my-s***"

validate_identifier

Utility for validating a single identifier such as a table or column name. As SQL injection prevention, only alphanumeric characters and underscores are permitted. Used automatically within DataAccess, but can also be used directly when building custom queries.

def validate_identifier(identifier: str, identifier_type: str = "identifier") -> str

Returns

strReturns the validated identifier unchanged; raises IdentifierValidationError for invalid input

identifier_type is a label for the identifier kind used in error messages and defaults to "identifier".

validate_identifiers

Utility for validating a list of identifiers (column names, table names, etc.) in a single batch. As SQL injection prevention, only alphanumeric characters and underscores are permitted.

def validate_identifiers(identifiers: List[str], identifier_type: str = "column") -> List[str]

Returns

List[str]Returns the validated identifiers in input order; raises IdentifierValidationError if any identifier is invalid

identifier_type names the kind of identifier being validated and defaults to "column". Note that this default differs from the singular validate_identifier (which defaults to "identifier").

validate_order_by

Utility for validating an ORDER BY clause. Only the column_name ASC|DESC format is allowed.

def validate_order_by(order_by: str) -> str

Returns

strReturns the validated ORDER BY clause unchanged; raises IdentifierValidationError for invalid input

ValidationPython

from agenticstar_platform.common import validate_identifier, validate_order_by

table = validate_identifier("users")        # OK: "users"
validate_identifier("users; DROP TABLE --")  # -> IdentifierValidationError

order = validate_order_by("created_at DESC") # OK: "created_at DESC"

AsyncContextManagerMixin

Mixin class for async context managers. Used internally by the SDK to uniformly implement the async with pattern.

class AsyncContextManagerMixin: async def __aenter__(self) -> Self: ... async def __aexit__(...) -> None: ... @abstractmethod async def close(self) -> None: ...

IdentifierValidationError

Exception raised by validate_identifier / validate_order_by for invalid identifiers. Inherits directly from Exception.

class IdentifierValidationError(Exception): ...

---

Error Handling

All modules use a common exception handling pattern. Each module has its own base exception class, with subclasses to distinguish specific errors.

Exception Hierarchy

Exception ├── IdentifierValidationError ├── VectorStoreError │ ├── VectorStoreConfigError │ │ └── QdrantConfigError │ └── VectorStoreConnectionError ├── EmbeddingError │ └── RateLimitExceededError ├── StorageError │ ├── StorageConfigError │ ├── StorageConnectionError │ └── StorageOperationError ├── AuthError │ ├── AuthConfigError │ └── AuthAPIError │ ├── AuthUnauthorizedError (401) │ ├── AuthNotFoundError (404) │ └── AuthRateLimitError (429) ├── SemanticMemoryError │ └── SemanticMemoryConfigError └── SecurityError ├── SecurityConfigError └── SecurityAPIError

Common Pattern

Across all modules, catch exceptions in order: specific exception -> base exception -> Exception.

Multi-Module Error HandlingPython

from agenticstar_platform.storage import StorageError, StorageOperationError
from agenticstar_platform.auth import AuthError, AuthAPIError
from agenticstar_platform.security import SecurityError, SecurityAPIError

async def process_document(doc_id: str):
    try:
        # 1. Retrieve document from DB
        doc = await db.fetch_one(
            "SELECT * FROM documents WHERE id = $1", (doc_id,)
        )

        # 2. Security check
        moderation = await security.check_content_moderation(doc["content"])
        if moderation.blocked:
            raise ValueError("Content policy violation")

        # 3. Generate Embedding + save to vector DB
        await qdrant.upsert(
            id=doc_id,
            content=doc["content"],
            metadata={"source": "documents"}
        )

        # 4. Save processing result to storage
        await storage.upload_bytes(
            data=doc["content"].encode(),
            object_name=f"processed/{doc_id}.txt"
        )

    except StorageOperationError as e:
        # Storage operation error -> determine by error_code
        logger.error(f"Storage op failed: {e.error_code}")
    except AuthAPIError as e:
        # Auth API error -> determine by status_code
        logger.error(f"Auth failed: {e.status_code}")
    except (StorageError, AuthError, SecurityError) as e:
        # Module base exception fallback
        logger.error(f"Module error: {type(e).__name__}: {e}")
    except Exception as e:
        # Unexpected error
        logger.critical(f"Unexpected: {e}")
        raise

---

metering Metering Module

v0.5.15+ — Pure infrastructure for measuring LLM usage & cost. Computes cost from an LLM response/usage and records one row per call to llm_usage_ledger, with daily rollup (llm_usage_daily) and cost-visualization queries. Agent-logic agnostic — identical for self-hosted and Marketplace BYO (in-process; no proxy/header coupling).

• Pluggable cost: defaults to litellm (completion_cost / cost_per_token), reflecting long-context / prompt-cache / tier pricing. When litellm is absent, tokens are still recorded (cost_usd = NULL). Override with cost_fn=.
• DB: pass any handle exposing execute_query(query, params) -> {success, data, error} (the SDK's DataAccess / PostgreSQLManager).

UsageMeter

Python

from agenticstar_platform.metering import UsageMeter

meter = UsageMeter(db=data_access)                  # cost_fn=, currency= optional
await meter.ensure_schema()                          # idempotently create ledger/daily/rollup and add newer columns to an existing ledger (skip if migrations already applied them)

# Record one LLM call (cost auto-computed; record failures are swallowed = safe)
await meter.record(
    model="gpt-5.5", response=resp, endpoint="chat/completions",
    labels={"execution_id": eid, "message_id": mid, "conversation_id": cid, "user_id": uid},
)

# Wrap a call so it records on completion
resp = await meter.track(model="gpt-5.5", labels=ids)(litellm.acompletion)(**params)

# Cost only (no record)
usd = UsageMeter.cost_usd("gpt-5.5", response=resp)

# Daily rollup + cost visualization (dashboards / billing)
await meter.rollup_recent()
rows = await meter.daily_cost(by="model", since_days=30)   # by = "model" | "user" | "agent" | "day"

Methods

Method	Description
UsageMeter(db, *, cost_fn=None, currency="USD")	Initialize with a DB handle and cost function
UsageMeter.cost_usd(model, response=None, usage=None, cost_fn=None)	Compute cost (USD) only (static)
await ensure_schema(*, with_rollup=True)	Idempotently create llm_usage_ledger / llm_usage_daily / rollup_llm_usage_daily()
await record(*, model, response=None, usage=None, endpoint=None, agent_type=None, level=None, labels=None, stream=False, success=True, request_id=None, cost_usd=None, missing_reason=None, fail_safe=True)	Record one row, return computed cost (missing_reason = reason when usage is absent; present rows store NULL. v0.5.21+)
track(*, model=None, endpoint=None, labels=None, ...)	Decorator wrapping an async LLM call for auto-recording
await rollup_recent()	Roll up previous & current day into the daily table (idempotent)
await daily_cost(*, by="model", since_days=30)	Fetch cost breakdown from the daily table
default_cost_fn(model, response=None, usage=None)	Module-level function. The default cost function used by UsageMeter when cost_fn is not supplied. Computes cost (USD) via litellm; returns None if it cannot be computed
extract_usage(usage)	Module-level function. Extracts {"input_tokens", "output_tokens", "total_tokens"} from a usage object/dict; returns None if no tokens can be found. When the prompt-cache breakdown is available it also returns cache_read_tokens / cache_creation_tokens (v3.0.0+; the keys are omitted when unknown)
round_cost_usd(v)	Module-level function. Formats a cost (USD) as a Decimal rounded to 8 decimal places

The default llm_usage_ledger DDL is portable (any PostgreSQL). Each row carries usage_status (present/missing) and missing_reason (reason when absent; v0.5.21+), so cost_usd IS NULL (unpriced), usage_status='missing', and true-zero are distinguishable. From v3.0.0, prompt-cache reads and writes are recorded separately in cache_read_tokens / cache_creation_tokens (both nullable; cached_tokens is unchanged). Reads and writes are priced very differently, so a merged figure cannot reveal a cache that is being written but never read. Explicit breakdown values (including 0) take precedence; otherwise the existing cached_tokens (reads) and cache_creation_input_tokens (writes) are used. If you pass a merged figure in cached_tokens, supply the breakdown explicitly. The daily rollup table and function are unchanged; aggregate the breakdown from the ledger. record() in v3.0.0 writes these two columns, so before you start writing, add them with ensure_schema() or a migration (the same ADD COLUMN IF NOT EXISTS), confirm the columns exist, and after upgrading confirm that rows are being saved (a database-side failure is not always raised as an exception). At scale, partition it monthly (e.g. pg_partman).

---

audit Audit Module

v0.5.31+ — Pure infrastructure that records agent-side action events (approvals/rejections, tool authorizations, policy violations, kill-switch, external-agent invocations) into append-only ledgers (agent_action_audit / llm_gateway_action_audit). Distinct from metering by its write guarantees — this is an audit trail, not telemetry. No extra dependencies (stdlib only, included in the core install).

• Callers pass no raw content (usage contract): never hand prompts / tool arguments to the ledger — one-way them with canonical_digest() (SHA-256) into payload_digest. The SDK performs no automatic secret detection or redaction. Field caps (reason truncated to 300 chars / payload_digest must be 64 lowercase-hex chars or it is dropped / metadata over 2KB of key-sorted JSON is replaced by its digest + size, always deep-copied) bound the blast radius of accidental leakage — they are not leak prevention by themselves.
• Three write modes: record() = at-least-once while the process lives (bounded retry → on final failure the sanitized row is spilled as JSON to a log line tagged action_audit_spill for manual reconcile) / record_sync() = audit-before-act (raises ActionAuditWriteError on DB write failure — an approval that cannot be audited must not take effect) / record_nowait() = fire-and-forget for hot paths (strong task refs, backlog capped at 512 — overflow spills instead of blocking; pre-start cancellation also spills; best-effort — lost if the process dies abruptly).
• Decision events require actor_ref: approval.* / gate.resolved / content.deleted / forensic.* / policy.changed / item.quarantine_released / workitem.* raise ValueError synchronously without it (v1.0.0+; before that only approval.*).
• DB: any handle exposing execute_query(query, params) -> {success, data, error} (the SDK's DataAccess / PostgreSQLManager).

ActionAudit / GatewayActionAudit

Python

from agenticstar_platform.audit import ActionAudit, GatewayActionAudit, canonical_digest

audit = ActionAudit(db=data_access, source="my-agent", system_version=image_tag)  # source = name of the writing component
await audit.ensure_schema()                       # once at provisioning, as the owner role: create ledger + indexes, add newer columns to older tables

# Hot path (tool authorization etc.): fire-and-forget
audit.record_nowait(
    event_type="tool.access", actor_type="agent", decision="deny",
    resource="bash", action="execute", reason="blocked_command",
    conversation_id=cid, execution_id=eid,
    payload_digest=canonical_digest({"command": cmd}),
)

# Approval (audit-before-act): abort the action if this raises
await audit.record_sync(
    event_type="approval.granted", actor_type="human", actor_ref=approver_id,
    decision="granted", resource=tool_name, conversation_id=cid,
)

await audit.aclose()                              # shutdown: drain pending writes (timeout → cancel + spill)

gw = GatewayActionAudit(db=gateway_da)            # LLM Gateway policy decisions (virtual-key identity)
await gw.record(event_type="policy.mcp_stripped", decision="stripped",
                virtual_key_id=vk, request_id=rid)

Methods

Method	Description
ActionAudit(db, *, source, system_version=None, agent_ref=None)	Writer for agent_action_audit. source (the name of the writing component; any string, stored in the row's source column) is required. agent_ref is the identifier filled in as actor_ref on rows with actor_kind='agent' that have none (default agent:<source>; v1.0.0+)
GatewayActionAudit(db, *, system_version=None)	Writer for llm_gateway_action_audit (virtual-key identity)
await ensure_schema()	Idempotently create ledger + indexes, then REVOKE UPDATE/DELETE (best-effort; use a dedicated insert-only role for production append-only enforcement). For ActionAudit it also adds the four v1.0.0 columns to tables created by older SDKs (ACTION_AUDIT_UPGRADE_DDL). CREATE / ALTER need the table owner's privileges, so call it once as the owner role when provisioning the environment (the runtime writer needs only INSERT). Do not call it in environments whose tables are managed by migrations. Existing duplicate source_event_id values block the UNIQUE index (deduplicate first)
await record(**fields)	At-least-once write. Returns True when the DB write succeeded, False when only spilled (validation ValueError raises synchronously; cancellation propagates)
await record_sync(**fields)	Audit-before-act write. Raises ActionAuditWriteError on DB write failure (validation raises ValueError)
record_nowait(**fields)	Fire-and-forget (returns an asyncio.Task; while closing / over backlog / without a loop it spills immediately and returns None)
await aclose(timeout=10.0)	Drain pending writes. Rows cancelled by the timeout are spilled; new events after closing spill immediately
canonical_digest(payload)	Module function. SHA-256 digest of any payload (key-sorted JSON, non-JSON types stringified; None stays None)
reference_keys(**refs)	Module function. Builds the dict of correlation keys for metadata (project_id / run_id / step_id / item_id / tool_call_id / root_execution_id etc.). Drops None and empty strings and stringifies values (v1.0.0+)
actor_kind_from_type(actor_type)	Module function. Derives actor_kind from actor_type (agent / assistant / worker → agent, human / user / approver → human, system / scheduler / policy → system, anything else → unknown; v1.0.0+)
KNOWN_EVENT_TYPES	Constant. Dict of known event_type names and their meanings (descriptive; writers do not enforce it; v1.0.0+)
ACTOR_KINDS	Constant. Closed vocabulary of actor_kind: ("agent", "human", "system", "human_external", "unknown") (v1.0.0+)
ACTION_AUDIT_DDL / ACTION_AUDIT_INDEXES / ACTION_AUDIT_UPGRADE_DDL	Constants. The DDL used by ensure_schema() (for environments without migrations; v1.0.0+)

Main fields: event_type / actor_type (recommended: human | agent | system; the implementation only checks non-empty) / actor_kind (closed vocabulary ACTOR_KINDS; derived from actor_type when omitted; values outside the vocabulary raise ValueError; v1.0.0+) / actor_ref (mandatory for the decision events above) / on_behalf_of_ref / decision / reason / resource / action / policy_id / policy_version / correlation keys (execution_id / conversation_id / message_id / request_id; v1.0.0+ also project_id (UUID string) / root_execution_id) / source_event_id (deduplication key, ^[A-Za-z0-9_.:@+-]{1,160}$; v1.0.0+) / payload_digest / metadata. Every row gets an automatic row_hash (SHA-256) for content-integrity verification.

Columns added in v1.0.0: actor_kind / project_id / root_execution_id / source_event_id are all optional, and existing calls (with actor_type only) keep working. When writing to an older table without these columns, the writer detects the missing column once and switches to the INSERT with only the existing columns, so no audit rows are lost. While it does, the values of the four new columns are not stored and row_hash is computed over the columns actually stored (it switches back to the new INSERT after ensure_schema() adds the columns, or after a process restart once the columns exist). On a table with the UNIQUE index on source_event_id, writing a duplicate value makes record() return False (spilled) and record_sync() raise.

metadata numbers and row_hash: numbers in metadata other than integers with an absolute value below 2^53 (floats, larger integers, NaN / Inf) are stored as strings (v1.0.0+), so that row_hash is identical whichever language computes it. If you write to the same table directly without the SDK, follow agenticstar_platform/audit/ROW_HASH_SPEC.md (the canonicalization rules) and row_hash_fixture.json (the verification fixture) shipped in the package. metadata that cannot be turned into JSON (for example because of circular references) is replaced with {"unserializable": true}, and the row itself is still saved.

GuardrailAlertWriter / GuardrailAlertAggregates (v0.5.32+)

Producer writers for the Admin triage view guardrail_alerts (non-authoritative — the authority remains the action-audit ledger) and the unlinked daily aggregates guardrail_alert_daily_aggregates. Delivery contract matches the writers above (bounded FAF + retry → spill → aclose() drain).

Python

from agenticstar_platform.audit import GuardrailAlertWriter, GuardrailAlertAggregates

alerts = GuardrailAlertWriter(db=data_access, system_version=image_tag)
await alerts.ensure_schema()                       # rows + aggregates tables + indexes + REVOKE
alerts.record_nowait(                              # confirmed blocks only; idempotent (event_key v2)
    source_surface="llm_gateway", policy_outcome="blocked",
    policy_source="llm_gateway_guardrails_settings", decision_point="input",
    request_id=rid, detection_types=["harmful_content"],
    moderation_categories={"Violence": 4},         # **measured** per-category severity (2/4/6)
)
agg = GuardrailAlertAggregates(db=data_access)
agg.record_nowait(source_surface="llm_gateway", category="SelfHarm", severity=6)

Structurally enforced: no end-user identity column on rows / putting SelfHarm on a row raises ValueError synchronously (normalized variants, nesting, and str-Enum .value are all detected; SelfHarm is aggregate-only and the aggregates writer takes no correlation-id arguments) / row policy_outcome is blocked-only / replays never clobber Admin lifecycle state (ON CONFLICT (event_key) DO NOTHING). Grants: rows = producer INSERT-only; aggregates = INSERT + UPDATE(count) + SELECT(count). Aggregate counts are approximate (at-least-once may double-count) — never use them for billing or regulatory reporting.

v0.5.40+: source_surface gains agent_worker (the conversation-pod worker surface). Rows gain policy_category, a closed vocabulary of attack techniques (prompt_injection / system_prompt_extraction / implementation_access / credential_request / file_system_access / security_bypass / custom_policy / unclassified) — out-of-vocabulary values raise ValueError synchronously (no free text or rule names on rows), and the SelfHarm scan also covers prompt_shield_origin / policy_category.

---

workflow Workflow Module

v1.0.1+ — The runner for agents that run once per step of a project workflow (protocol mp-workflow/1; Marketplace edition 3.1 and later; because the result format changed in v3.0.1, the Marketplace edition 3.1 requires v3.0.1 or later). It is a separate entry point from run_marketplace_agent for chat. Included in core and needs no database driver (YAML files declared in the contract are parsed strictly and validated with the core dependency pyyaml). See the Workflow Guide for how to write an agent.

• Runs only on the injected environment variables: ASTER_RUNTIME_PROTOCOL / ASTER_RUNTIME_BASE_URL / ASTER_RUNTIME_TOKEN / EXECUTION_ID. No fallback to DB settings.
• Parent and child processes: the runner (parent) fetches and stages the inputs, records the start, collects and saves the outputs and saves the result, and runs your agent function in a child process.
• Exit codes: 0 = the result was saved (including when the run had already ended, failed/agent_error when your agent raised or the entry point could not be imported, and when the platform refused the result after a cancellation (v3.0.13+)), 2 = it could not be saved, 3 = the environment variables or the execution context are invalid. Other failures may end with a non-zero exit and the exception (not guaranteed).

Entry points

def run(entrypoint: str, *, env: Optional[Mapping[str, str]]= None, workspace: Optional[str]= None, exit_process: bool = True, stop_on_sigterm: Optional[bool]= None) -> WorkflowResult: ... # stop_on_sigterm: v3.0.13+ async def arun(entrypoint: str, *, env: Optional[Mapping[str, str]]= None, workspace: Optional[str]= None, **kwargs) -> WorkflowResult: ... def validate_env(env: Mapping[str, str]) -> Dict[str, str]: ...

Function / constant	Description
run(entrypoint, *, env=None, workspace=None, exit_process=True, stop_on_sigterm=None)	The entry point called from the image's start command. entrypoint is "module.path:function". With exit_process=True (default) the process exits with the exit code, and on SIGTERM (pod stop) it switches to a cooperative stop that stops the child processes and collects the outputs. With exit_process=False it returns the WorkflowResult and raises WorkflowConfigError for invalid configuration. stop_on_sigterm (v3.0.13+) sets the cooperative stop on SIGTERM separately from exit_process (default: the value of exit_process). From v3.0.13 it also makes the runner's token unreadable from your agent's processes through /proc, and adds a log output (standard error, only when logging is not configured)
await arun(entrypoint, ...)	Async version. Does not install signal handlers, token protection or a log output (pass stop_on_sigterm=True / protect_token=True to enable them)
validate_env(env)	Validates the required environment variables (missing or malformed values raise WorkflowConfigError)
PROTOCOL / REQUIRED_ENV	"mp-workflow/1" / names of the required environment variables

WorkflowResult has execution_id / outcome / outcome_reason_code / had_mutating_success / completion_saved / already_terminal / exit_code / artifacts.

WorkflowContext

The read-only execution context passed to the entry point (async def run(context)).

class WorkflowContext: execution_id: str project_id: str run_id: str agent_id: str marketplace_agent_id: str root_execution_id: str entry_ref: str operation_key: str attempt: int effect_mode: str # read_only | idempotent_write | non_idempotent_write parameters: Mapping[str, Any]# read-only connections: Mapping[str, str]# connection slot name -> connection ID input_dir: Path# /workspace/input (read-only) work_dir: Path# /workspace/work output_dir: Path# /workspace/output manifest: Mapping[str, Any] deadlines: Mapping[str, Optional[str]] gate_decision: Optional[Mapping[str, Optional[str]]]# v3.0.9+ audit: AuditApi def deadline_remaining() -> Optional[float]: ... async def is_cancel_requested(*, max_age: float = 5.0) -> bool: ... async def ensure_can_act() -> None: ... async def skip(reason: str) -> None: ... async def policy_unavailable() -> None: ... async def log(message: str) -> None: ... async def connection_token(slot: str, *, min_ttl: float = 300.0) -> Mapping[str, Any]: ... # v3.0.5+

Method	Description
deadline_remaining()	Seconds left until your agent is stopped (None if unknown)
await is_cancel_requested(max_age=5.0)	True once a cancellation, deadline or stop is known. Within max_age seconds of the previous check it does not ask the runner and answers from the stops it already knows
await ensure_can_act()	Call before an external operation. Raises WorkflowError after cancellation or the deadline
await skip(reason)	Skips a step with nothing to process. reason is no_target / not_applicable / nothing_to_do (anything else raises ValueError). It only records the decision and does not stop your agent, so return from the function after calling it
await policy_unavailable()	Reports that whether an operation is allowed could not be checked before any external operation (skipped/policy_unavailable). It only records the decision, so return from the function after calling it. Calling it after a write has started results in failed/skip_after_effect
await log(message)	Writes to the runner log (the pod's standard error; up to 2000 characters)
await connection_token(slot, *, min_ttl=300.0)	Returns the connection's access token as {access_token, token_type, expires_at} (read-only). Refreshes it once less than min_ttl seconds remain, and every 600 seconds when the expiry is unknown. A slot not bound at startup, a revoked connection, or any call after cancellation or the deadline raises WorkflowError (v3.0.5+)

gate_decision is a read-only {gate_id, decision, decided_by, note, decided_at} only on a run started by a review decision (the step the work was sent back to, run again, or the step that follows an advance), and None on every other run. decision is send_back / proceed, decided_by is human / system, and note is the reason or comment the reviewer wrote (None when there is none or the decision was automatic). The start record (manifest) includes only gate_id / decision / decided_by (v3.0.9+; see the Workflow Guide).

AuditApi

The audit sender available as context.audit.

class AuditApi: async def tool_invoked(tool: str, *, effect_kind: str, action_ref: Optional[str]= None, target: Optional[str]= None, **extra) -> str: ... async def tool_effect(invoked_event_id: str, *, status: str = "ok", count: Optional[int]= None, action_ref: Optional[str]= None, **extra) -> str: ... async def tool_denied(tool: str, *, reason: str, reason_code: str = "tool_denied", effect_kind: str = "write", **extra) -> str: ...

Method	Description
await tool_invoked(tool, *, effect_kind, ...)	Call before an external operation. Returns the event ID after the record has been saved; raises AuditNotPersisted if it cannot be saved (do not proceed with the operation). effect_kind is read / list / query / fetch or write / create / update / delete / send / mutate (anything else raises ValueError)
await tool_effect(invoked_event_id, *, status="ok", ...)	Links the result of the operation to the return value of tool_invoked. status is ok / failed / unknown (use unknown when the result is not known = effect unknown)
await tool_denied(tool, *, reason, reason_code="tool_denied", ...)	Records that the operation was not run. The result becomes denied (conditions earlier in the evaluation order, such as cancellation and deadlines, take precedence)

Of the values passed in **extra, only strings, integers, finite floats and booleans are kept in the metadata (the SDK drops anything else and, from v3.0.13, logs the names of the dropped keys). Do not pass bodies, tokens or argument values.

Execution result

Name	Description
OUTCOMES	("completed", "incomplete", "failed", "timeout", "stopped", "denied", "skipped")
ENVELOPE_VERSION	Version of the saved result format (completion envelope). Currently 2 (v3.0.1+)
EffectTally / decide_outcome() / build_envelope()	Low-level API the runner uses to build the result. Your agent does not need to call them

The result's counts contain tool_calls / read_calls / mutating_invoked / mutating_success / mutating_failed / mutating_unresolved / orphan_effects / effect_conflicts / denied / audit_persist_failures (tool_calls / read_calls were renamed from tools_invoked / reads in v3.0.1). For the list of outcomes and reason codes, see the Workflow Guide.

Exceptions

WorkflowError (RuntimeError) ├── WorkflowConfigError # injected environment variables missing or mismatched, or a context for another execution / protocol ├── RuntimeApiError # runtime API response outside the contract (status / code / message) ├── ExecutionTerminal # the execution is already terminal (never a reason to rerun the agent) ├── AuditNotPersisted # an audit record could not be saved (including after cancellation or the deadline; do not proceed) ├── InputIntegrityError # input file digest / size / path mismatch └── OutputCollectionError # contract violation while collecting outputs (path / size / regular file / symlink)

connection_token() / ensure_can_act() and similar calls also raise WorkflowError after cancellation or the deadline. From v3.0.6, calls made after the runner has closed the control channel also raise WorkflowError (runner closed the control channel).

---

policy Policy Module

v1.0.0+ — Pure computation that decides, as a binary allow / forbid, whether a tool call is permitted under the combination of the tenant's governance policy and the project's policy. No I/O and no framework dependency; included in core.

• Two layers: the tenant-level L1Policy (default profile, an unremovable floor, whether rationale bodies are retained, version) and the project-level ProjectPolicy (profile, deny / allow, gate settings). tighten_merge() guarantees that the floor cannot be removed by a project, that the profile can never be looser than the tenant default, and that a strict allow list only works within the operation classes the tenant profile permits.
• Profiles: open (everything allowed) < guarded (destructive forbidden) < strict (reads allowed; writes / destructive only when listed in allow).
• Evaluation order: no policy → forbid / matches the floor or deny → forbid / MCP or A2A call → allow (mcp_unrestricted / a2a_unrestricted; v3.0.3+ / v3.0.4+) / arguments cannot be interpreted → forbid / unclassified tool → allow under open / guarded (unregistered=True), forbid under strict, delegation or dry_run / classified → the profile table (dry_run=True forbids everything but reads).
• Forbid rules: no scope = a tool-name glob (MCP calls are matched against both mcp:<connection name>:<tool> and <tool>), optionally with argument conditions (args) and an operation-class condition (op_class). op_class conditions only apply to built-in tools (MCP and A2A calls are not classified by operation class, so they are never matched; to forbid MCP methods, use an external rule with a connection ID and method glob, or a tool-name glob). internal = an operation kind of built-in tools (kind + mapping_version). external = a connection ID + method glob (connection_kind is mcp / a2a; A2A rules apply per connection and have no tool). External connections are matched by ID, not by name (v3.0.2+).
• Storing and verifying: EffectivePolicy.to_dict() / from_dict() store and restore a policy, and digest (SHA-256) lets you check later which settings a decision was made under. The stored schema_version is 4 when A2A rules exist, 3 when scoped rules exist, and 2 otherwise.

Evaluation examplePython

from agenticstar_platform.policy import L1Policy, ProjectPolicy, tighten_merge, evaluate

l1 = L1Policy.from_row({"version": 3, "default_profile": "guarded", "floor": {"forbid": ["mcp:*:delete_*"]}})
project = ProjectPolicy.from_json({"profile": "strict", "allow": ["write_file"]})
policy = tighten_merge(l1, project, project_id="p1")

evaluate(policy, "delete_repo", None, mcp=True, mcp_server="gitlab", connection_id="12")  # forbid (floor)
evaluate(policy, "create_issue", None, mcp=True, mcp_server="gitlab", connection_id="12") # allow (mcp_unrestricted)
evaluate(policy, "write_file", "write")        # allow (strict_allowlist)
evaluate(policy, "delete_file", "destructive")  # forbid (strict_allowlist)
policy.digest                                   # SHA-256 of the settings used for the decision

Functions and classes

Name	Description
tighten_merge(l1, project, *, project_id=None)	Combines the two layers and returns an EffectivePolicy (when l1 is None, the tenant baseline is open with no floor; project restrictions still apply)
evaluate(policy, tool, op_class, *, args=None, args_unparseable=False, dry_run=False, delegated=False, mcp_server=None, connection_id=None, mcp=None, connection_kind=None, tool_aliases=None)	Returns a Decision. op_class is the caller's classification: read / write / destructive (None when unknown). For MCP pass mcp=True and connection_id; for A2A pass connection_kind="a2a" and connection_id. If the connection cannot be identified while external rules of that kind exist, the call is forbidden. For in-process MCP servers that are not connections registered in the admin console, pass LOCAL_CONNECTION ("local"; not subject to external rules)
Decision	mode (allow / forbid) / reason / policy_version / unregistered / matched (the matching forbid pattern) / layer (v3.0.11+; for a forbid, which rule layer decided it: project = the project's settings (deny rules, profile, allowlist, dry run) can change it, tenant = the tenant's default profile, floor = the tenant's floor, which nobody can relax, system = not a rule (no policy, or the arguments could not be evaluated); None for an allow; constants LAYER_PROJECT / LAYER_TENANT / LAYER_FLOOR / LAYER_SYSTEM), plus the allowed property
L1Policy.from_row(row) / ProjectPolicy.from_json(raw)	Build the inputs from a DB row / JSON
EffectivePolicy.to_dict() / EffectivePolicy.from_dict(raw) / .digest	Store, restore and SHA-256. from_dict() rejects an unsupported schema_version with ValueError (callers treat it as forbid-all)
policy_digest(policy)	SHA-256 of an EffectivePolicy
verify_connections(policy, connections)	Checks that the connections referenced by rules exist (ValueError otherwise). connections is {"mcp": {...}, "a2a": {...}}, or an MCP map / set (v3.0.2+; per kind since v3.0.4)
requires_schema(floor, forbid)	The schema_version the rules require
profile_rank() / tighten_profile() / profile_permits(profile, op_class)	Profile order, the stricter of two, and whether a profile alone can permit an operation class
internal_operation_kinds(mapping_version=1)	IDs, labels, descriptions and version of the internal operation kinds (text_write / file_read / file_transfer / command_exec / browser / desktop), in a form safe to show in UIs

Reason codes: policy_unavailable / floor / project_forbid / unevaluable_args / unknown_tool / dry_run_read_only / profile / strict_allowlist / mcp_unrestricted / a2a_unrestricted. Constants: PROFILE_ORDER / CONNECTION_KINDS (("mcp", "a2a")) / LOCAL_CONNECTION ("local") / SCHEMA_VERSION_SCOPED (3) / SCHEMA_VERSION_TYPED_CONNECTION (4) / SUPPORTED_SCHEMA_VERSIONS / INTERNAL_OPERATIONS_MAPPING_VERSION.

---

Summary

AGENTIC STAR Platform SDK v3.0.15 is a comprehensive toolkit consisting of the following 13 specialized modules:

• Events — Async event delivery (SSE / Webhook / DB)
• Database — PostgreSQL + Azure AD auth + specialized access layers (Config / Execution / Telemetry / PodRuntime)
• RAG — Embedding + Vector Search (Qdrant)
• Storage — Cloud Storage (Azure / S3 / GCS)
• Auth — AGENTIC STAR authentication service integration
• Memory — Semantic Memory (Mem0 + Qdrant)
• Security — Content Moderation + PII Detection (Azure / AWS / GCP multi-cloud support)
• Common — Common utilities (SQL injection prevention, secret masking)
• Metering — LLM usage & cost metering (ledger + daily rollup, v0.5.15+)
• Runner — Marketplace-compatible execution lifecycle (run_marketplace_agent: identity validation → input fetch → run → result persistence/webhook → exactly-one terminal → cleanup, v0.5.30+)
• Audit — Action-audit ledger (ActionAudit / GatewayActionAudit: append-only record of approvals / tool authorizations / policy violations; callers pass SHA-256 digests instead of raw content, v0.5.31+)
• Workflow — Runner for agents that run in steps of a project workflow (run / WorkflowContext: stage inputs → run → collect outputs → save the result, audit of external operations, cancellation and deadlines, connection tokens, v1.0.1+)
• Policy — Tool-call governance policy (tighten_merge / evaluate: combines the tenant and project layers and decides allow / forbid, v1.0.0+)

Each module can be installed and used independently, and by combining them you can build powerful AI agent applications. The SDK provides infrastructure only, and developers freely design agent logic.