apcore-toolkit-python

Python implementation of the apcore-toolkit.

Extracts ~1,400 lines of duplicated framework-agnostic logic from django-apcore and flask-apcore into a standalone Python package.

Installation

pip install apcore-toolkit

Core Modules

Usage

Scanning and Writing

from apcore_toolkit import BaseScanner, ScannedModule, YAMLWriter

class MyScanner(BaseScanner):
    def scan(self, **kwargs):
        # Scan your framework endpoints and return ScannedModule instances
        return [
            ScannedModule(
                module_id="users.get_user",
                description="Get a user by ID",
                input_schema={"type": "object", "properties": {"id": {"type": "integer"}}, "required": ["id"]},
                output_schema={"type": "object", "properties": {"name": {"type": "string"}}},
                tags=["users"],
                target="myapp.views:get_user",
            )
        ]

    def get_source_name(self):
        return "my-framework"

scanner = MyScanner()
modules = scanner.scan()

# Filter and deduplicate
modules = scanner.filter_modules(modules, include=r"^users\.")
modules = scanner.deduplicate_ids(modules)

# Write YAML binding files
writer = YAMLWriter()
writer.write(modules, output_dir="./bindings")

Direct Registry Registration

from apcore import Registry
from apcore_toolkit import RegistryWriter

registry = Registry()
writer = RegistryWriter()
writer.write(modules, registry)

Conformance Verification

Approval and ACL gating key on a module's requires_approval annotation, which is only reachable if annotations survive scan → register → get_definition. A writer that drops them disables that gate silently. Run the shared conformance check in your adapter's test suite so any such regression fails loudly:

from apcore import ModuleAnnotations, Registry
from apcore_toolkit import assert_annotations_preserved

def test_writer_preserves_annotations():
    module = my_scanned_module(  # a ScannedModule with a resolvable target
        annotations=ModuleAnnotations(destructive=True, requires_approval=True),
    )
    # Raises AssertionError if annotations were dropped or changed.
    assert_annotations_preserved(MyRegistryWriter(), module, Registry())

Output Format Factory

from apcore_toolkit.output import get_writer

writer = get_writer("yaml")    # YAMLWriter
writer = get_writer("python")  # PythonWriter
writer = get_writer("registry")  # RegistryWriter

Pydantic Model Flattening

from apcore_toolkit import flatten_pydantic_params, resolve_target

# Resolve a target string to a callable
func = resolve_target("myapp.views:create_task")

# Flatten Pydantic model params into scalar kwargs for MCP tools
wrapped = flatten_pydantic_params(func)

OpenAPI Schema Extraction

from apcore_toolkit.openapi import extract_input_schema, extract_output_schema

input_schema = extract_input_schema(operation, openapi_doc)
output_schema = extract_output_schema(operation, openapi_doc)

Schema Enrichment

from apcore_toolkit import enrich_schema_descriptions

enriched = enrich_schema_descriptions(schema, {"user_id": "The user ID"})

Markdown Formatting

from apcore_toolkit import to_markdown

md = to_markdown({"name": "Alice", "role": "admin"}, title="User Info")

Surface-Aware Formatters (v0.6.0)

format_module / format_modules / format_schema render ScannedModule and JSON Schema for specific consumer surfaces — LLM context, agent skill files, CLI listings, or programmatic JSON.

from apcore_toolkit import format_module, format_modules, format_schema

# Module styles: "markdown" (default), "skill", "table-row", "json"
md = format_module(module, style="markdown")
skill_file = format_module(module, style="skill")        # ---\nname: ...\ndescription: ...\n---
row = format_module(module, style="table-row")           # CLI listing row
payload = format_module(module, style="json")            # dict (for APIs)

# Batch with optional grouping by tag or module-id prefix
listing = format_modules(modules, style="markdown", group_by="tag")

# Schema styles: "prose" (default), "table", "json"
prose = format_schema(schema, style="prose", max_depth=3)
table = format_schema(schema, style="table")

See apcore-toolkit/docs/features/formatting.md for the full contract.

Tabular Formats (v0.7.0)

Byte-equivalent CSV / JSONL emitters with a cross-SDK conformance contract — Python, TypeScript, and Rust produce identical bytes for the same input.

from apcore_toolkit import format_csv, format_jsonl

rows = [
    {"sn": 1, "title": "First", "score": 78},
    {"sn": 2, "title": "Second", "score": 82, "description": "later-only field"},
]

# CSV: header = union of keys across all rows (no silent data loss on
# heterogeneous rows); nested values serialized as canonical compact JSON;
# RFC 4180 CRLF line terminator.
csv_text = format_csv(rows)
# 'sn,title,score,description\r\n1,First,78,\r\n2,Second,82,later-only field\r\n'

# JSONL: canonical compact JSON per row, LF terminator, no trailing blank.
jsonl_text = format_jsonl(rows)

# UTF-8 BOM for Excel locales (default off for pipeline consumers):
csv_for_excel = format_csv(rows, bom=True)

See apcore-toolkit/docs/features/formatting.md § Tabular Formats for the full contract and apcore-toolkit/conformance/fixtures/format_csv.json / format_jsonl.json for the shared cross-SDK test corpus.

Display Overlay (§5.13)

DisplayResolver applies a sparse binding.yaml display overlay to a list of ScannedModule instances, populating metadata["display"] with surface-facing presentation fields (alias, description, guidance, tags) for CLI, MCP, and A2A surfaces.

from apcore_toolkit.display import DisplayResolver

resolver = DisplayResolver()

# Apply overlay from a directory of *.binding.yaml files
modules = resolver.resolve(scanned_modules, binding_path="bindings/")

# Or from a pre-parsed dict
modules = resolver.resolve(
    scanned_modules,
    binding_data={
        "bindings": [
            {
                "module_id": "product.get",
                "display": {
                    "alias": "product-get",
                    "description": "Get a product by ID",
                    "cli": {"alias": "get-product"},
                    "mcp": {"alias": "get_product"},
                },
            }
        ]
    },
)

# Resolved fields are in metadata["display"]
mod = modules[0]
print(mod.metadata["display"]["cli"]["alias"])   # "get-product"
print(mod.metadata["display"]["mcp"]["alias"])   # "get_product"
print(mod.metadata["display"]["a2a"]["alias"])   # "product-get"

Resolution chain (per field): surface-specific override > display default > binding-level field > scanner value.

MCP alias constraints: automatically sanitized (non-[a-zA-Z0-9_-] chars replaced with _; leading digit prefixed with _); raises ValueError if result exceeds 64 characters.

CLI alias validation: warns and falls back to display.alias when a user-explicitly-set alias does not match ^[a-z][a-z0-9_-]*$.

Convention Module Discovery (§5.14)

ConventionScanner scans a directory of plain Python files for public functions and converts them to ScannedModule instances. No decorators, no base classes, no imports from apcore -- just functions with type hints.

from apcore_toolkit import ConventionScanner

scanner = ConventionScanner()
modules = scanner.scan(commands_dir="commands/")

# Each public function becomes a module:
#   commands/deploy.py  ->  deploy.deploy
#   commands/deploy.py  ->  deploy.rollback  (if rollback() exists)

Module-level constants customize behavior:

# commands/deploy.py
MODULE_PREFIX = "ops"       # override file-based prefix -> ops.deploy
CLI_GROUP = "operations"    # group hint for CLI surface
TAGS = ["infra", "deploy"]  # tags stored in metadata

def deploy(env: str, tag: str = "latest") -> dict:
    """Deploy the app to the given environment."""
    return {"status": "deployed", "env": env}

Input and output schemas are inferred from PEP 484 type annotations. Use include / exclude regex filters to control which module IDs are registered.

Documentation

Full documentation is available at https://github.com/aiperceivable/apcore-toolkit.

License

Apache-2.0