Skip to main content

Your First Agent

This guide walks you through creating and running a minimal custom agent using the agenticstar-platform SDK. The SDK provides infrastructure (DB, events, storage, etc.), while you freely design the agent logic.

Prerequisites​

  • Python 3.12 or higher
  • pip or uv
  • PostgreSQL database (local or Azure Database for PostgreSQL)
  • Azure OpenAI API key (for LLM calls)

Step 1: Install the SDK​

curl
pip install agenticstar-platform[all]==0.5.3

Step 2: Create the Configuration File​

Create a config.toml file. Each SDK module reads its configuration from this file.

config.tomlToml
[postgresql]
host = "localhost"
port = 5432
database = "agenticstar"
user = "agent_user"
password = "your_password"
min_connections = 2
max_connections = 10

[postgresql.azure_ad]
enabled = false

[events]
enable_db_handler = true
enable_webhook_handler = false

Step 3: Implement the Agent​

Create my_agent.py. This example implements a simple agent that retrieves job information using the SDK's DB module and streams progress via the Events module.

my_agent.pyPython
import asyncio
from openai import AsyncAzureOpenAI

from agenticstar_platform.db import PostgreSQLManager, PostgreSQLConfig, DataAccess
from agenticstar_platform.db.execution_access import ExecutionAccess
from agenticstar_platform.events import EventEmitter, EventType

# --- Configuration ---
db_config = PostgreSQLConfig.from_toml("config.toml")
llm_client = AsyncAzureOpenAI(
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    api_version="2024-12-01-preview",
    api_key="your-api-key",
)

async def run_agent(execution_id: str, user_message: str):
    """Minimal agent execution flow"""

    # 1. Initialize DB connection
    db_manager = PostgreSQLManager(db_config)
    da = DataAccess(db_manager)
    await da.initialize()

    execution_access = ExecutionAccess(da)

    # 2. Initialize EventEmitter
    emitter = EventEmitter(execution_id=execution_id)

    # 3. Emit execution start event
    await emitter.emit_event(
        event_type=EventType.PHASE_START,
        message="Processing request...",
        metadata={"phase": "processing"},
    )

    # 4. Retrieve past messages
    history = await execution_access.get_messages(execution_id=execution_id)

    # 5. Call LLM (agent logic is freely designed)
    messages = [{"role": "system", "content": "You are a helpful assistant."}]
    if history:
        for h in history:
            messages.append({"role": h["role"], "content": h["content"]})
    messages.append({"role": "user", "content": user_message})

    response = await llm_client.chat.completions.create(
        model="gpt-4.1",
        messages=messages,
    )
    answer = response.choices[0].message.content

    # 6. Emit completion event
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message=answer,
    )

    # 7. Cleanup
    await emitter.cleanup()
    await da.close()

    return answer

if __name__ == "__main__":
    result = asyncio.run(run_agent("exec-001", "Hello!"))
    print(result)

Step 4: Verify​

curl
python my_agent.py

If everything works correctly, the LLM response will be printed to the console.

Step 5: Add Tools (Optional)​

Here is an example of adding tools (function calling) to your agent. This uses the SDK's RAG module to search a knowledge base.

tools/search_knowledge.pyPython
from agenticstar_platform.rag import QdrantManager, QdrantConfig
from agenticstar_platform.rag import EmbeddingGenerator, EmbeddingConfig

async def search_knowledge(query: str, top_k: int = 5) -> list[dict]:
    """Search the knowledge base for documents related to the query"""

    # Generate embedding
    embed_config = EmbeddingConfig.from_toml("config.toml")
    generator = EmbeddingGenerator(embed_config)
    query_vector = await generator.generate(query)

    # Similarity search with Qdrant
    qdrant_config = QdrantConfig.from_toml("config.toml")
    qdrant = QdrantManager(qdrant_config)

    results = await qdrant.search(
        collection_name="knowledge_base",
        query_vector=query_vector,
        limit=top_k,
    )

    return [
        {"content": r.payload.get("content", ""), "score": r.score}
        for r in results
    ]

Example Project Structure​

my-agent/
├── config.toml              # SDK configuration file
├── my_agent.py              # Agent main logic
├── tools/
│   ├── search_knowledge.py  # RAG search tool
│   └── file_manager.py      # Storage operations tool
├── Dockerfile               # Container build
├── requirements.txt
└── README.md

Next Steps​