SDK API Reference

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

Installation

Install the SDK using pip.

Basic Installationcurl
pip install agenticstar-platform==0.5.29

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]==0.5.29

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

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

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

Module List

The SDK consists of the following 10 modules.

Module	Description	Key Classes
events	Async event delivery system (SSE / Webhook / DB)	EventEmitter, StreamingEvent
db	PostgreSQL connection + Azure AD auth	PostgreSQLManager, DataAccess
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.29+)	run_marketplace_agent

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]: ...

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]: ...

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.

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.

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()

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']}")

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}")

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.

class MCPTokenInfo(BaseModel): access_token: str token_type: str expires_at: datetime 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 ) -> 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 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}")

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

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()

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

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.

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 (skip if already managed)

# 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
`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. At scale, partition it monthly (e.g. pg_partman).

Summary

AGENTIC STAR Platform SDK v0.5.29 is a comprehensive toolkit consisting of the following 10 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.29+)

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.

Installation​

Module List​

events Events Module​

Enums​

EventType​

SubEventType​

Data Classes​

StreamingEvent​

SequencedEvent​

ExecutionMessage​

Classes​

EventEmitter​

DatabaseEventHandler​

WebhookEventHandler​

CompositeEventHandler​

Factory Functions​

Protocol​

EventHandler​

db Database Module​

Configuration​

AzureADConfig​

PostgreSQLConfig​

Connection Management​

PostgreSQLManager​

Data Access Layer​

DataAccess​

ConfigAccess​

ExecutionAccess​

TelemetryAccess​

PodRuntime​

ApiPostgreSQLManager​

Factory Function​

Error Handling​

rag RAG Module​

Enums​

VectorStoreProvider​

Exceptions​

Data Classes​

SearchResult​

UpsertResult​

SearchResponse​

Embedding Configuration​

EmbeddingConfig​

EmbeddingGenerator​

Vector Database​

QdrantConfig​

PayloadIndexConfig​

VectorStoreClientBase​

VectorStoreClientProtocol​

QdrantManager​

storage Storage Module​

Enums​

StorageProvider​

Exceptions​

Result Types​

UploadResult​

DownloadResult​

ObjectInfo​

ListResult​

Configuration​

StorageConfig​

AzureBlobConfig​

S3Config​

GCSConfig​

Storage Clients​

AzureBlobStorageClient​

S3StorageClient​

GCSStorageClient​

auth Auth Module​

Exceptions​

Configuration​

AgenticStarAuthConfig​

Data Models​

ApiUser​

DeviceInfo​

MCPTokenInfo​

UserPagination​

LoginHistoryEntry​

MCPTokenError​

OAuthProviderName​

Installation

Module List

events Events Module

Enums

EventType

SubEventType

Data Classes

StreamingEvent

SequencedEvent

ExecutionMessage

Classes

EventEmitter

DatabaseEventHandler

WebhookEventHandler

CompositeEventHandler

Factory Functions

Protocol

EventHandler

db Database Module

Configuration

AzureADConfig

PostgreSQLConfig

Connection Management

PostgreSQLManager

Data Access Layer

DataAccess

ConfigAccess

ExecutionAccess

TelemetryAccess

PodRuntime

ApiPostgreSQLManager

Factory Function

Error Handling

rag RAG Module

Enums

VectorStoreProvider

Exceptions

Data Classes

SearchResult

UpsertResult

SearchResponse

Embedding Configuration

EmbeddingConfig

EmbeddingGenerator

Vector Database

QdrantConfig

PayloadIndexConfig

VectorStoreClientBase

VectorStoreClientProtocol

QdrantManager

storage Storage Module

Enums

StorageProvider

Exceptions

Result Types

UploadResult

DownloadResult

ObjectInfo

ListResult

Configuration

StorageConfig

AzureBlobConfig

S3Config

GCSConfig

Storage Clients

AzureBlobStorageClient

S3StorageClient

GCSStorageClient

auth Auth Module

Exceptions

Configuration

AgenticStarAuthConfig

Data Models

ApiUser

DeviceInfo

MCPTokenInfo

UserPagination

LoginHistoryEntry

MCPTokenError

OAuthProviderName