naas-abi-core
The core implementation library for ABI (Agentic Brain Infrastructure), providing the fundamental building blocks for building unified AI systems. This library implements the core concepts, services, and architecture patterns that enable ontology-driven AI applications.
Overview
The ABI Library is the core implementation of ABI's concepts, designed to build a unified AI system. This library provides the fundamental building blocks for connecting, processing, and utilizing data across different AI components.
naas-abi-core is the foundational library that powers the ABI framework. It provides:
- Engine: Central orchestration system that loads and coordinates modules, services, and ontologies
- Services: Hexagonal architecture-based services for storage, secrets, and AI capabilities
- Modules: Modular system for organizing agents, integrations, pipelines, and workflows
- Agents: LangGraph-based AI agents with tool binding and conversation management
- Applications: Ready-to-use interfaces (REST API, Terminal, MCP Protocol)
- NATS mode (3.0): the engine serves its kernel services on a NATS broker, the bus is JetStream, module jobs replace Dagster, and modules can run in their own process with
naas-abi-sdk
Upgrading from 2.x: Migrate from 2.x to 3.0 and Breaking changes in 3.0.
Installation
pip install naas-abi-core
Optional Dependencies
# NATS mode: naas-abi-proto, naas-abi-sdk, nats-py (needed by naas-abi)
pip install naas-abi-core[nats]
# OpenTelemetry tracing
pip install naas-abi-core[otel]
# Qdrant vector store, AWS S3 object storage, SSH tunnels, Redis, OpenRouter
pip install naas-abi-core[qdrant,aws,ssh,redis,openrouter]
# Every supported extra (nats and otel included)
pip install naas-abi-core[all]
# Legacy, for projects without NATS, removed in the next major: Dagster, RabbitMQ
pip install 'naas-abi-core[all,dagster,rabbitmq]'
naas-abi-core[all] does not include the legacy dagster and rabbitmq
extras. Without them, DagsterOrchestration.New(), the RabbitMQ bus adapter
and the Dagster app raise MissingLegacyExtra, which prints the install
command.
NATS mode
A top-level nats: block in the config switches the engine to NATS mode:
nats:
nats_url: "nats://abi:{{ secret.NATS_ABI_PASSWORD }}@127.0.0.1:4222"
jwt_secret: "{{ secret.NATS_JWT_SECRET }}"
discovery:
project: "my_project"
- Every kernel service is served on
abi.svc.<service>.v1.*(protobuf contracts fromnaas-abi-proto), and modules reach services through NATS facades, even in one process. - The bus is NATS JetStream; module jobs (
naas_abi_core.module.jobs:@job,Cron,Every,OnEvent) run scheduled and event-driven work. - One engine serves per NATS account (
nats.engine.role,ABI_ROLLOUT_IDfor handovers); discovery registers remote modules. - Agent memory and the activity log live in the Document Service.
Architecture: NATS and the Kernel Services, Remote Modules and Jobs.
Core Architecture
Engine
The Engine is the central orchestrator that:
- Loads Configuration: Reads and validates YAML configuration files
- Initializes Services: Sets up storage, vector, triple store, and secret services based on module dependencies
- Loads Modules: Discovers and loads modules with their agents, integrations, pipelines, and workflows
- Loads Ontologies: Loads RDF ontologies into the triple store for semantic reasoning
- Initializes Components: Calls
on_initialized()on all modules after everything is loaded
Example Usage:
from naas_abi_core.engine.Engine import Engine
# Initialize engine with default configuration (config.yaml)
engine = Engine()
# Load all modules
engine.load()
# Or load specific modules
engine.load(module_names=["naas_abi", "my_custom_module"])
# Access loaded modules
for module_name, module in engine.modules.items():
print(f"Module: {module_name}")
print(f"Agents: {[agent.__name__ for agent in module.agents]}")
# Access services
triple_store = engine.services.triple_store
vector_store = engine.services.vector_store
object_storage = engine.services.object_storage
secret_service = engine.services.secret
Modules
Modules are the primary organizational unit in ABI. Each module can contain:
- Agents: AI agents that can be used for conversations and task execution
- Integrations: Connections to third-party services and APIs
- Pipelines: Data transformation processes that convert raw data into semantic representations
- Workflows: Business logic that can be exposed as tools, API endpoints, or scheduled jobs
- Ontologies: RDF/Turtle files that define semantic knowledge structures
Module Structure:
from naas_abi_core.module.Module import BaseModule, ModuleConfiguration
from naas_abi_core.engine.EngineProxy import EngineProxy
class MyModule(BaseModule):
class Configuration(ModuleConfiguration):
# Module-specific configuration
api_key: str
dependencies = ModuleDependencies(
modules=[], # Other modules this module depends on
services=[TripleStoreService, VectorStoreService] # Required services
)
def on_load(self):
# Called when module is loaded
# Load ontologies, agents, etc.
super().on_load()
def on_initialized(self):
# Called after all modules and services are initialized
# Safe to access other modules and services here
pass
Services
Services form the foundational layer of ABI, implementing the Hexagonal Architecture (Ports & Adapters) pattern to provide flexible and system-agnostic interfaces. This architectural approach allows ABI to seamlessly integrate with existing systems while maintaining clean separation of concerns.
Each service defines a primary port (interface) that specifies its capabilities, while multiple secondary adapters can implement this interface for different backend systems. This means you can:
- Easily swap implementations without changing business logic
- Add new integrations by implementing new adapters
- Test components in isolation using mock adapters
For example, the Secret Service could connect to various backend systems through different adapters:
- Hashicorp Vault
- AWS Secrets Manager
- Azure Key Vault
- Environment Variables
- Local File System
- Google Cloud Secret Manager
- Kubernetes Secrets
This modular approach ensures that ABI can be deployed in any environment while maintaining consistent interfaces and behavior across different infrastructure choices.
Triple Store Service
Manages RDF knowledge graphs for semantic reasoning and ontology storage.
Supported Backends:
- Oxigraph (default)
- SPARQL endpoints
- Custom adapters
Example:
from naas_abi_core.services.triple_store.TripleStoreService import TripleStoreService
triple_store = engine.services.triple_store
# Query the knowledge graph
results = triple_store.query("""
SELECT ?subject ?predicate ?object
WHERE {
?subject ?predicate ?object
}
LIMIT 10
""")
Vector Store Service
Manages vector embeddings for semantic search and similarity matching.
Supported Backends:
- Qdrant (optional, requires
[qdrant]extra) - Custom adapters
Example:
from naas_abi_core.services.vector_store.VectorStoreService import VectorStoreService
vector_store = engine.services.vector_store
# Store embeddings
vector_store.upsert(
collection_name="intents",
vectors=[embedding],
ids=["intent_1"],
payloads=[{"text": "user query"}]
)
# Search similar vectors
results = vector_store.search(
collection_name="intents",
query_vector=query_embedding,
limit=5
)
Object Storage Service
Manages file storage for documents, reports, and generated content.
Supported Backends:
- AWS S3 (optional, requires
[aws]extra) - MinIO
- Local file system
- Custom adapters
Example:
from naas_abi_core.services.object_storage.ObjectStorageService import ObjectStorageService
object_storage = engine.services.object_storage
# Upload a file
object_storage.upload(
bucket="my-bucket",
key="documents/report.pdf",
file_path="/path/to/report.pdf"
)
# Download a file
object_storage.download(
bucket="my-bucket",
key="documents/report.pdf",
file_path="/path/to/downloaded.pdf"
)
Secret Service
Manages secrets and credentials securely across different storage systems.
Supported Backends:
- Environment variables
- Naas Secret Manager
- Hashicorp Vault
- AWS Secrets Manager
- Azure Key Vault
- Google Cloud Secret Manager
- Kubernetes Secrets
- Local file system
- Custom adapters
Example:
from naas_abi_core.services.secret.Secret import Secret
secret_service = engine.services.secret
# Get a secret
api_key = secret_service.get("OPENAI_API_KEY")
# List all secrets
all_secrets = secret_service.list()
Cache Service
Provides intelligent caching for API calls, tool results, and model responses to optimize performance and manage rate limits.
Example:
from naas_abi_core.services.cache.CacheService import CacheService
cache = CacheService()
# Cache a function result
@cache.cache(ttl=3600)
def expensive_api_call(param: str):
# Expensive operation
return result
# Force refresh
result = expensive_api_call(param, force_refresh=True)
Core Concepts
Integration
Integrations provide standardized connections to third-party services and data sources. They handle:
- Authentication and authorization
- API communication
- Data format standardization
- Error handling and retries
Example:
from naas_abi_core.integration.integration import Integration, IntegrationConfiguration
class MyAPIConfiguration(IntegrationConfiguration):
api_key: str
base_url: str
class MyAPIIntegration(Integration):
def __init__(self, configuration: MyAPIConfiguration):
super().__init__(configuration)
# Initialize connection
def fetch_data(self):
# Implement API call
pass
Pipeline
Pipelines are responsible for data ingestion and transformation into the ontological layer. They:
- Utilize integrations to fetch data
- Transform raw data into semantic representations
- Maintain data consistency and quality
- Map external data models to ABI's ontology
Example:
from naas_abi_core.pipeline.pipeline import Pipeline, PipelineConfiguration, PipelineParameters
from rdflib import Graph
class MyPipelineConfiguration(PipelineConfiguration):
integration_config: dict
class MyPipelineParameters(PipelineParameters):
source: str
class MyPipeline(Pipeline):
def __init__(self, configuration: MyPipelineConfiguration):
super().__init__(configuration)
def run(self, parameters: MyPipelineParameters) -> Graph:
# Fetch data from integration
# Transform to RDF
# Return Graph
graph = Graph()
# ... add triples to graph
return graph
def trigger(self, event, ontology_name, triple) -> Graph:
# Event-driven pipeline execution
return self.run(MyPipelineParameters(source="event"))
Workflow
Workflows leverage the ontological layer to implement business logic and provide data to consumers. They can be used by:
- Large Language Models (LLMs)
- Remote APIs and services
- Other automated processes
Example:
from naas_abi_core.workflow.workflow import Workflow, WorkflowConfiguration, WorkflowParameters
from pydantic import BaseModel
class MyWorkflowParameters(WorkflowParameters):
input_data: str
class MyWorkflowConfiguration(WorkflowConfiguration):
processing_option: str
class MyWorkflow(Workflow[MyWorkflowParameters]):
def __init__(self, configuration: MyWorkflowConfiguration):
super().__init__(configuration)
def run(self, parameters: MyWorkflowParameters):
# Implement business logic
# Query knowledge graph
# Process data
# Return results
return {"result": "processed"}
Agent
Agents are AI-powered assistants that can have conversations, use tools, and delegate to sub-agents.
Features:
- LangGraph-based conversation management
- Tool binding and execution
- Sub-agent delegation
- Intent-based routing
- Conversation persistence (PostgreSQL checkpointing)
- Event streaming for real-time updates
Example:
from naas_abi_core.services.agent.Agent import Agent
from langchain_openai import ChatOpenAI
class MyAgent(Agent):
def __init__(self):
super().__init__(
name="MyAgent",
description="An agent that helps with specific tasks",
system_prompt="You are a helpful assistant...",
chat_model=ChatOpenAI(model="gpt-4"),
tools=[my_tool, my_workflow],
agents=[sub_agent] # Optional sub-agents
)
Applications
REST API
FastAPI-based REST API that automatically exposes all agents, workflows, and pipelines.
Features:
- OpenAPI/Swagger documentation
- OAuth2 authentication
- CORS support
- Automatic endpoint generation from agents/workflows/pipelines
Usage:
from naas_abi_core.apps.api.api import api
# Run the API server
api()
Endpoints:
GET /- API landing pageGET /docs- Swagger UI documentationGET /redoc- ReDoc documentationPOST /agents/{agent_name}/completion- Agent completion endpointPOST /workflows/{workflow_name}/run- Workflow execution endpointPOST /pipelines/{pipeline_name}/run- Pipeline execution endpoint
Terminal Agent
Interactive terminal interface for chatting with agents.
Usage:
from naas_abi_core.apps.terminal_agent.main import run_agent
from naas_abi_core.services.agent.Agent import Agent
agent = MyAgent()
run_agent(agent)
MCP Server
Model Context Protocol (MCP) server for integration with Claude Desktop and VS Code.
Features:
- Dynamic agent discovery from OpenAPI spec
- HTTP and stdio transport modes
- Automatic tool registration
Usage:
# Start MCP server
python -m naas_abi_core.apps.mcp.mcp_server
# Or with HTTP transport
MCP_TRANSPORT=http python -m naas_abi_core.apps.mcp.mcp_server
Configuration
ABI uses YAML configuration files (typically config.yaml) to configure:
- Services: Storage backends, connection details
- Modules: Which modules to load and their configurations
- API: API title, description, CORS settings
- Global Config: AI mode (cloud, local, airgap)
Example Configuration:
api:
title: "My ABI API"
description: "API for my AI system"
cors_origins:
- "http://localhost:9879"
global_config:
ai_mode: "cloud" # or "local" or "airgap"
nats:
nats_url: "nats://abi:{{ secret.NATS_ABI_PASSWORD }}@127.0.0.1:4222"
jwt_secret: "{{ secret.NATS_JWT_SECRET }}"
services:
triple_store:
triple_store_adapter:
adapter: "oxigraph"
config:
oxigraph_url: "http://localhost:7878"
vector_store:
vector_store_adapter:
adapter: "qdrant"
config:
host: "localhost"
port: 6333
object_storage:
object_storage_adapter:
adapter: "fs"
config:
base_path: "storage/datastore"
secret:
secret_adapters:
- adapter: "dotenv"
config: {}
modules:
- module: "naas_abi"
enabled: true
- path: "./src/my_module"
enabled: true
config:
api_key: "{{ secret.MY_API_KEY }}"
The full reference, with every service and the nats:, telemetry: and
dev: blocks: Configuration Reference.
Key Dependencies
- rdflib: RDF and ontology management
- langgraph: Agent conversation management
- langgraph-checkpoint-postgres: Conversation persistence (NATS off)
- naas-abi-sdk, naas-abi-proto, nats-py (
[nats]extra): NATS mode - fastapi: REST API framework
- sparqlwrapper: SPARQL query execution
- pydantic: Data validation and configuration
- loguru: Logging
- langchain-openai: OpenAI integration
Architecture Patterns
Hexagonal Architecture (Ports & Adapters)
All services follow the Hexagonal Architecture pattern:
- Primary Port: Interface defining service capabilities
- Secondary Adapters: Implementations for different backends
- Benefits: Easy swapping of implementations, testability, system-agnostic design
Module System
- Dependency Resolution: Automatic dependency resolution and ordering
- Lazy Loading: Services loaded only when needed by modules
- Lifecycle Hooks:
on_load()andon_initialized()for setup - Isolation: Modules can be developed and tested independently
Event-Driven Pipelines
Pipelines can be triggered by:
- Manual execution via API or code
- Ontology events (triple insertions/updates)
- Scheduled jobs
- Workflow triggers
Development
Running Tests
pytest
Type Checking
mypy naas_abi_core
Building
uv build
See Also
- ABI Main README - Complete ABI framework documentation
- naas-abi-cli - CLI tool for ABI projects
- naas-abi-marketplace - Marketplace modules and agents
License
MIT License
Documentation
1 maintainer
Metadata
1 maintainer
Release files for naas-abi-core 3.0.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| naas_abi_core-3.0.0.tar.gz | 1.7 MB | With provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| naas_abi_core-3.0.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 3.5 MB
Release files / naas_abi_core-3.0.0.tar.gz
| Download URL | naas_abi_core-3.0.0.tar.gz |
|---|---|
| Size | 1.7 MB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
ad6acaf0ea266c9574d08d6a4246cec489ac30a1fca51dc8dd050225485d6aa3
|
|
BLAKE2b-256 checksum How to use checksums |
c92e199fe08e1bfb9d7f52ea53643f4566a5cedb0f60d1ca23604648a58a240a
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 20, 2026.
Transparency logRelease files / naas_abi_core-3.0.0-py3-none-any.whl
| Download URL | naas_abi_core-3.0.0-py3-none-any.whl |
|---|---|
| Size | 1.8 MB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
e168cfea8edb37d8f3ba0b6ea82d95ecdf780fe3ebb66fee9fca10bd118179d1
|
|
BLAKE2b-256 checksum How to use checksums |
5d7e5963592cf55d46733c6b89b27c88f17673c488d03247d1b0c2639ffd908e
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 20, 2026.
Transparency logSecurity for naas-abi-core 3.0.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
Full provenance
All 2 files in this release have been uploaded with provenance metadata.
Attestation for naas_abi_core-3.0.0.tar.gz
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 20, 2026.
Transparency logA note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
