apcore

Schema-driven module development framework for AI-perceivable interfaces.

apcore provides a unified task orchestration framework with strict type safety, access control, middleware pipelines, and built-in observability. It enables you to define modules with structured input/output schemas that are easily consumed by LLMs and other automated systems.

Features

• Schema-driven modules -- Define input/output contracts using Pydantic models with automatic validation
• 10-step execution pipeline -- Context creation, safety checks, ACL enforcement, validation, middleware chains, and execution with timeout support
• @module decorator -- Turn plain functions into fully schema-aware modules with zero boilerplate
• YAML bindings -- Register modules declaratively without modifying source code
• Access control (ACL) -- Pattern-based, first-match-wins rules with wildcard support
• Middleware system -- Composable before/after hooks with error recovery
• Observability -- Tracing (spans), metrics collection, and structured context logging
• Async support -- Seamless sync and async module execution
• Safety guards -- Call depth limits, circular call detection, frequency throttling

Requirements

• Python >= 3.11

Installation

pip install -e .

For development:

pip install -e ".[dev]"

Quick Start

Define a module with the decorator

from apcore import module

@module(description="Add two integers", tags=["math"])
def add(a: int, b: int) -> int:
    return a + b

Define a module with a class

from pydantic import BaseModel
from apcore import Context

class GreetInput(BaseModel):
    name: str

class GreetOutput(BaseModel):
    message: str

class GreetModule:
    input_schema = GreetInput
    output_schema = GreetOutput
    description = "Greet a user"

    def execute(self, inputs: dict, context: Context) -> dict:
        return {"message": f"Hello, {inputs['name']}!"}

Register and execute

from apcore import Registry, Executor

registry = Registry()
registry.register("greet", GreetModule())

executor = Executor(registry=registry)
result = executor.call("greet", {"name": "Alice"})
# {"message": "Hello, Alice!"}

Add middleware

from apcore import LoggingMiddleware, TracingMiddleware

executor.use(LoggingMiddleware())
executor.use(TracingMiddleware())

Access control

from apcore import ACL, ACLRule

acl = ACL(rules=[
    ACLRule(callers=["admin.*"], targets=["*"], effect="allow", description="Admins can call anything"),
    ACLRule(callers=["*"], targets=["admin.*"], effect="deny", description="Others cannot call admin modules"),
])
executor = Executor(registry=registry, acl=acl)

Project Structure

src/apcore/
    __init__.py          # Public API
    context.py           # Execution context & identity
    executor.py          # Core execution engine
    decorator.py         # @module decorator
    bindings.py          # YAML binding loader
    config.py            # Configuration
    acl.py               # Access control
    errors.py            # Error hierarchy
    module.py            # Module annotations & metadata
    middleware/           # Middleware system
    observability/       # Tracing, metrics, logging
    registry/            # Module discovery & registration
    schema/              # Schema loading, validation, export
    utils/               # Utilities

Development

Run tests

pytest

Run tests with coverage

pytest --cov=src/apcore --cov-report=html

Lint and format

ruff check --fix src/ tests/
ruff format src/ tests/

Type check

mypy src/ tests/

📄 License

Apache-2.0

🔗 Links

• Documentation: docs/apcore - Complete documentation
• Website: aipartnerup.com
• GitHub: aipartnerup/apcore
• PyPI: apcore
• Issues: GitHub Issues
• Discussions: GitHub Discussions