ADP Python SDK

Python SDK for the Agentic Data Protocol (ADP) — a deterministic, policy-aware protocol that allows AI agents to access heterogeneous data systems safely and reproducibly via a unified intent interface.

This SDK provides a client-side API for connecting to any ADP-compliant server (such as adp-hypervisor) over stdio transport.

Installation

# Using uv (recommended)
uv add adp-sdk

# Using pip
pip install adp-sdk

Quick Start

import asyncio
from adp_sdk import stdio_client, basic_auth, QueryIntent, PredicateGroup

async def main():
    async with stdio_client(
        "python", ["-m", "adp_hypervisor", "--config", "config.yaml"],
        authorization=basic_auth("alice", "secret"),
    ) as session:
        # 1. Discover available resources
        discovery = await session.discover()
        for resource in discovery.resources:
            print(f"  {resource.resource_id}: {resource.description}")

        # 2. Describe a resource to get its usage contract
        contract = await session.describe("com.example:users", "QUERY")
        print(f"Fields: {[f.field_id for f in contract.usage_contract.fields]}")

        # 3. Execute an intent
        intent = QueryIntent(
            intentClass="QUERY",
            resourceId="com.example:users",
            predicates=PredicateGroup(op="AND", predicates=[]),
        )
        result = await session.execute(intent)
        print(f"Got {len(result.results)} rows")

asyncio.run(main())

API Overview

stdio_client (convenience)

async with stdio_client(command, authorization=basic_auth("user", "secret")) as session:
    ...

Launches an ADP server as a subprocess and returns an initialized ClientSession.

ClientSession

Method	Description
discover(filter?, cursor?)	List available resources
describe(resource_id, intent_class, version?, cursor?)	Get a resource's usage contract
validate(intent)	Check if an intent is valid
execute(intent, cursor?)	Run an intent against a resource
ping()	Check server liveness

Error Handling

All server errors are mapped to typed exceptions:

from adp_sdk import ResourceNotFoundError, ValidationFailedError

try:
    result = await session.execute(intent)
except ResourceNotFoundError as e:
    print(f"Resource not found: {e}")
except ValidationFailedError as e:
    print(f"Invalid intent: {e}")

Development

# Install dependencies
uv sync --all-extras

# Run tests
uv run python -m unittest discover -s tests -v

# Type checking
uv run pyright

# Linting
uv run ruff check .

License

Apache-2.0