apcore

Build once, invoke by Code or AI.

A schema-enforced module standard for the AI-Perceivable era.

apcore is an AI-Perceivable module standard that makes every interface naturally perceivable and understandable by AI through enforced Schema definitions and behavioral annotations. It provides strict type safety, access control, middleware pipelines, and built-in observability — enabling you to define modules with structured input/output schemas that are easily consumed by both code and AI.

Features

• Schema-driven modules -- Define input/output contracts using Pydantic models (with automatic validation) or plain JSON Schema dicts
• Execution Pipeline -- Context creation, safety checks, ACL enforcement, approval gate, 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
• Approval system -- Pluggable approval gate (Step 5) with sync/async handlers, Phase B resume, and audit events
• Extension points -- Unified extension management for discoverers, middleware, ACL, approval handlers, span exporters, and module validators
• Async task management -- Background module execution with status tracking, cancellation, and concurrency limiting
• Behavioral annotations -- Declare module traits (readonly, destructive, idempotent, cacheable, paginated, streaming) for AI-aware orchestration
• W3C Trace Context -- traceparent header injection/extraction for distributed tracing interop

API Overview

Core

Class	Description
APCore	High-level client -- register modules, call, stream, validate
Registry	Module storage -- discover, register, get, list, watch
Executor	Execution engine -- call with middleware pipeline, ACL, approval
Context	Request context -- trace ID, identity, call chain, cancel token
Config	Configuration -- load from YAML, get/set values
Identity	Caller identity -- id, type, roles, attributes
FunctionModule	Wrapped function module created by @module decorator

Access Control & Approval

Class	Description
ACL	Access control -- rule-based caller/target authorization
ApprovalHandler	Pluggable approval gate protocol
AlwaysDenyHandler / AutoApproveHandler / CallbackApprovalHandler	Built-in approval handlers

Middleware

Class	Description
Middleware	Pipeline hooks -- before/after/on_error interception
BeforeMiddleware / AfterMiddleware	Single-phase middleware adapters
LoggingMiddleware	Structured logging middleware
RetryMiddleware	Automatic retry with backoff
ErrorHistoryMiddleware	Records errors into ErrorHistory
PlatformNotifyMiddleware	Emits events on error rate/latency spikes
ObsLoggingMiddleware	Observability-aware structured logging middleware

Schema

Class	Description
SchemaLoader	Load schemas from YAML or native types
SchemaValidator	Validate data against schemas
SchemaExporter	Export schemas for MCP, OpenAI, Anthropic, generic
RefResolver	Resolve $ref references in JSON Schema

Observability

Class	Description
TracingMiddleware	Distributed tracing with span export
MetricsMiddleware / MetricsCollector	Call count, latency, error rate metrics
ContextLogger	Context-aware structured logging
ErrorHistory	Ring buffer of recent errors with deduplication
UsageCollector	Per-module usage statistics and trends
UsageMiddleware	Per-call usage tracking middleware
TraceContext	W3C Trace Context propagation (traceparent/tracestate)
InMemoryExporter	Span exporter that stores spans in memory
StdoutExporter	Span exporter that writes spans to stdout
OTLPExporter	Span exporter using OpenTelemetry Protocol

Events & Extensions

Class	Description
EventEmitter	Event system -- subscribe, emit, flush
WebhookSubscriber / A2ASubscriber	Built-in event subscribers
ExtensionManager	Unified extension point management
AsyncTaskManager	Background module execution with status tracking
CancelToken	Cooperative cancellation token
BindingLoader	Load modules from YAML binding files
ErrorCodeRegistry	Central registry for structured error codes

Documentation

For full documentation, including Quick Start guides for both Python and TypeScript, visit: https://aiperceivable.github.io/apcore/getting-started.html

Requirements

• Python >= 3.11

Installation

pip install apcore

Development

pip install -e ".[dev]"

Quick Start

Simple usage (Global Client)

For simple scripts or prototypes, you can use the global apcore functions:

import apcore

@apcore.module(id="math.add", description="Add two integers")
def add(a: int, b: int) -> int:
    return a + b

# Directly call it
result = apcore.call("math.add", {"a": 10, "b": 5})
print(result)  # {'result': 15}

Simplified Client (Recommended)

The APCore client provides a unified entry point that manages everything for you:

from apcore import APCore

client = APCore()

@client.module(id="math.add", description="Add two integers")
def add(a: int, b: int) -> int:
    return a + b

# Call the module
result = client.call("math.add", {"a": 10, "b": 5})
print(result)  # {'result': 15}

Advanced: Define a module with a class

from pydantic import BaseModel
from apcore import Context, APCore

client = APCore()

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

client.register("greet", GreetModule())
result = client.call("greet", {"name": "Alice"})
# {"message": "Hello, Alice!"}

Alternative: Define schemas with plain dicts

If you prefer not to use Pydantic, pass raw JSON Schema dicts directly:

from apcore import APCore

client = APCore()

class WeatherModule:
    input_schema = {"type": "object", "properties": {"city": {"type": "string"}}}
    output_schema = {"type": "object", "properties": {"temp": {"type": "number"}}}
    description = "Get current temperature"

    def execute(self, inputs: dict, context=None) -> dict:
        return {"temp": 22.5}

client.register("weather", WeatherModule())
result = client.call("weather", {"city": "Tokyo"})
# {"temp": 22.5}

Note: Dict schemas skip Pydantic input validation. Use Pydantic models when you need automatic type coercion and validation, or validate inside execute().

Add middleware

from apcore import LoggingMiddleware, TracingMiddleware

client.use(LoggingMiddleware())
client.use(TracingMiddleware())

Access control

from apcore import ACL, ACLRule, Executor, Registry

registry = Registry()
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)

Examples

The examples/ directory contains runnable demos:

---

simple_client — APCore client with decorator-based modules

Initializes an APCore client, registers modules with @client.module(), and calls them directly.

from apcore import APCore

client = APCore()

@client.module(id="math.add", description="Add two integers")
def add(a: int, b: int) -> int:
    return a + b

result = client.call("math.add", {"a": 10, "b": 5})
print(result)  # {'result': 15}

@client.module(id="greet")
def greet(name: str, greeting: str = "Hello") -> dict:
    return {"message": f"{greeting}, {name}!"}

result = client.call("greet", {"name": "Alice"})
print(result)  # {'message': 'Hello, Alice!'}

---

global_client — Minimal global client usage

No explicit initialization needed — use the default global client directly.

import apcore

@apcore.module(id="math.add")
def add(a: int, b: int) -> int:
    return a + b

result = apcore.call("math.add", {"a": 10, "b": 5})
print(result)  # {'result': 15}

---

greet — Duck-typed module with Pydantic schemas

Demonstrates the class-based module interface with Pydantic BaseModel for input/output schemas.

from pydantic import BaseModel

class GreetInput(BaseModel):
    name: str

class GreetOutput(BaseModel):
    message: str

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

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

---

get_user — Readonly module with ModuleAnnotations

Demonstrates behavioral annotations (readonly, idempotent) and simulated database lookup.

from pydantic import BaseModel
from apcore.module import ModuleAnnotations

class GetUserInput(BaseModel):
    user_id: str

class GetUserOutput(BaseModel):
    id: str
    name: str
    email: str

class GetUserModule:
    input_schema = GetUserInput
    output_schema = GetUserOutput
    description = "Get user details by ID"
    annotations = ModuleAnnotations(readonly=True, idempotent=True)

    _users = {
        "user-1": {"id": "user-1", "name": "Alice", "email": "alice@example.com"},
        "user-2": {"id": "user-2", "name": "Bob", "email": "bob@example.com"},
    }

    def execute(self, inputs: dict, context) -> dict:
        user_id = inputs["user_id"]
        user = self._users.get(user_id)
        if user is None:
            return {"id": user_id, "name": "Unknown", "email": "unknown@example.com"}
        return dict(user)

---

send_email — Destructive module with sensitive fields and ContextLogger

Shows x-sensitive on schema fields (for log redaction), ModuleAnnotations with metadata, ModuleExample for AI-perceivable documentation, and ContextLogger usage.

from pydantic import BaseModel, Field
from apcore.module import ModuleAnnotations, ModuleExample
from apcore.observability import ContextLogger

class SendEmailInput(BaseModel):
    to: str
    subject: str
    body: str
    api_key: str = Field(..., json_schema_extra={"x-sensitive": True})

class SendEmailOutput(BaseModel):
    status: str
    message_id: str

class SendEmailModule:
    input_schema = SendEmailInput
    output_schema = SendEmailOutput
    description = "Send an email message"
    tags = ["email", "communication", "external"]
    version = "1.2.0"
    metadata = {"provider": "example-smtp", "max_retries": 3}
    annotations = ModuleAnnotations(destructive=True, idempotent=False, open_world=True)
    examples = [
        ModuleExample(
            title="Send a welcome email",
            inputs={"to": "user@example.com", "subject": "Welcome!", "body": "...", "api_key": "sk-xxx"},
            output={"status": "sent", "message_id": "msg-12345"},
            description="Sends a welcome email to a new user.",
        ),
    ]

    def execute(self, inputs: dict, context) -> dict:
        logger = ContextLogger.from_context(context, name="send_email")
        logger.info("Sending email", extra={"to": inputs["to"], "subject": inputs["subject"]})
        message_id = f"msg-{hash(inputs['to']) % 100000:05d}"
        logger.info("Email sent successfully", extra={"message_id": message_id})
        return {"status": "sent", "message_id": message_id}

---

decorated_add — @module decorator for simple functions

from apcore.decorator import module

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

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: aiperceivable.com
• GitHub: aiperceivable/apcore-python
• PyPI: apcore
• Issues: GitHub Issues
• Discussions: GitHub Discussions