APIP SDK Core

Python interfaces for writing gateway policies on the WSO2 API Platform.

A policy is a plain Python class that the gateway executor loads at runtime. It receives a rich, fully-typed execution context and returns a structured action that the gateway applies to the live request or response—no configuration files, no generated stubs, no runtime dependencies.

---

Contents

• Requirements
• Installation
• Quick start
• Policy types
  • RequestHeaderPolicy
  • RequestPolicy
  • ResponseHeaderPolicy
  • ResponsePolicy
  • StreamingRequestPolicy
  • StreamingResponsePolicy
  • FaultPolicy
• Processing mode
• Actions reference
• Context types
• Versioned imports
• Development
• License

---

Requirements

• Python 3.10 or newer
• No third-party runtime dependencies

---

Installation

pip install apip-sdk-core

---

Quick start

1 · Add a request header

from typing import Any
from apip_sdk_core import (
    ExecutionContext,
    HeaderProcessingMode,
    ProcessingMode,
    RequestHeaderAction,
    RequestHeaderContext,
    RequestHeaderPolicy,
    UpstreamRequestHeaderModifications,
)


class AddCorrelationId(RequestHeaderPolicy):
    """Injects an X-Correlation-ID header into every upstream request."""

    def mode(self) -> ProcessingMode:
        return ProcessingMode(
            request_header_mode=HeaderProcessingMode.PROCESS,
        )

    def on_request_headers(
        self,
        execution_ctx: ExecutionContext,
        ctx: RequestHeaderContext,
        params: dict[str, Any],
    ) -> RequestHeaderAction:
        return UpstreamRequestHeaderModifications(
            headers_to_set={"x-correlation-id": execution_ctx.request_id},
        )

2 · Validate and rewrite a buffered request body

import json
from typing import Any
from apip_sdk_core import (
    BodyProcessingMode,
    ExecutionContext,
    ImmediateResponse,
    ProcessingMode,
    RequestAction,
    RequestContext,
    RequestPolicy,
    UpstreamRequestModifications,
)


class EnforceJsonSchema(RequestPolicy):
    """Rejects requests that are not valid JSON objects."""

    def mode(self) -> ProcessingMode:
        return ProcessingMode(request_body_mode=BodyProcessingMode.BUFFER)

    def on_request_body(
        self,
        execution_ctx: ExecutionContext,
        ctx: RequestContext,
        params: dict[str, Any],
    ) -> RequestAction:
        raw = ctx.body.content if ctx.body else None
        try:
            data = json.loads(raw or b"{}")
        except json.JSONDecodeError:
            return ImmediateResponse(
                status_code=400,
                body=b'{"error":"request body must be valid JSON"}',
                headers={"content-type": "application/json"},
            )
        # Optionally rewrite the body before forwarding
        return UpstreamRequestModifications(body=json.dumps(data).encode())

3 · Stream and log response chunks

import logging
from typing import Any
from apip_sdk_core import (
    BodyProcessingMode,
    ExecutionContext,
    ForwardResponseChunk,
    ProcessingMode,
    ResponseContext,
    ResponseAction,
    ResponseStreamContext,
    StreamBody,
    StreamingResponseAction,
    StreamingResponsePolicy,
    UpstreamRequestModifications,
)

logger = logging.getLogger(__name__)


class StreamingLogger(StreamingResponsePolicy):
    """Logs every streamed response chunk and passes it through unchanged."""

    def mode(self) -> ProcessingMode:
        return ProcessingMode(response_body_mode=BodyProcessingMode.STREAM)

    def needs_more_response_data(self, accumulated: bytes) -> bool:
        return False  # dispatch every chunk immediately

    def on_response_body(
        self,
        execution_ctx: ExecutionContext,
        ctx: ResponseContext,
        params: dict[str, Any],
    ) -> ResponseAction:
        return None  # fallback – never called in STREAM mode

    def on_response_body_chunk(
        self,
        execution_ctx: ExecutionContext,
        ctx: ResponseStreamContext,
        chunk: StreamBody,
        params: dict[str, Any],
    ) -> StreamingResponseAction:
        logger.debug(
            "chunk #%d  eos=%s  bytes=%d",
            chunk.index,
            chunk.end_of_stream,
            len(chunk.chunk),
        )
        return ForwardResponseChunk(body=chunk.chunk)

---

Policy types

All policy interfaces live in apip_sdk_core (re-exported from apip_sdk_core.policy.v1alpha2).

Interface	Hook called	Typical use
RequestHeaderPolicy	on_request_headers	Auth, routing, header injection
RequestPolicy	on_request_body	Body validation, transformation
ResponseHeaderPolicy	on_response_headers	CORS, response header rewriting
ResponsePolicy	on_response_body	Response body transformation
StreamingRequestPolicy	on_request_body_chunk	Streaming body inspection
StreamingResponsePolicy	on_response_body_chunk	Streaming response inspection / SSE
FaultPolicy	on_fault	Fault handling: notify, record, reshape the error

Every interface extends the base Policy class:

class Policy(ABC):
    @abstractmethod
    def mode(self) -> ProcessingMode: ...

    def close(self) -> None: ...   # optional – release resources

RequestHeaderPolicy

Called before the request body is read.

class RequestHeaderPolicy(Policy, ABC):
    @abstractmethod
    def on_request_headers(
        self,
        execution_ctx: ExecutionContext,
        ctx: RequestHeaderContext,
        params: dict[str, Any],
    ) -> RequestHeaderAction: ...

Returns UpstreamRequestHeaderModifications | ImmediateResponse | None

RequestPolicy

Called once with the complete buffered request body.

class RequestPolicy(Policy, ABC):
    @abstractmethod
    def on_request_body(
        self,
        execution_ctx: ExecutionContext,
        ctx: RequestContext,
        params: dict[str, Any],
    ) -> RequestAction: ...

Returns UpstreamRequestModifications | ImmediateResponse | None

ResponseHeaderPolicy

Called before the response body is read.

class ResponseHeaderPolicy(Policy, ABC):
    @abstractmethod
    def on_response_headers(
        self,
        execution_ctx: ExecutionContext,
        ctx: ResponseHeaderContext,
        params: dict[str, Any],
    ) -> ResponseHeaderAction: ...

Returns DownstreamResponseHeaderModifications | ImmediateResponse | None

ResponsePolicy

Called once with the complete buffered response body.

class ResponsePolicy(Policy, ABC):
    @abstractmethod
    def on_response_body(
        self,
        execution_ctx: ExecutionContext,
        ctx: ResponseContext,
        params: dict[str, Any],
    ) -> ResponseAction: ...

Returns DownstreamResponseModifications | ImmediateResponse | None

StreamingRequestPolicy

Extends RequestPolicy. Called chunk-by-chunk for the request body.

class StreamingRequestPolicy(RequestPolicy, ABC):
    @abstractmethod
    def needs_more_request_data(self, accumulated: bytes) -> bool: ...

    @abstractmethod
    def on_request_body_chunk(
        self,
        execution_ctx: ExecutionContext,
        ctx: RequestStreamContext,
        chunk: StreamBody,
        params: dict[str, Any],
    ) -> StreamingRequestAction: ...

Returns ForwardRequestChunk | None

StreamingResponsePolicy

Extends ResponsePolicy. Called chunk-by-chunk for the response body.

class StreamingResponsePolicy(ResponsePolicy, ABC):
    @abstractmethod
    def needs_more_response_data(self, accumulated: bytes) -> bool: ...

    @abstractmethod
    def on_response_body_chunk(
        self,
        execution_ctx: ExecutionContext,
        ctx: ResponseStreamContext,
        chunk: StreamBody,
        params: dict[str, Any],
    ) -> StreamingResponseAction: ...

Returns ForwardResponseChunk | TerminateResponseChunk | None

FaultPolicy

Called when a request fails, over the error response the gateway produced. A policy that implements this is attached through faultPolicies (globalFaultPolicies on the LLM kinds) rather than the normal chain, and never runs on a successful response.

class FaultPolicy(Policy, ABC):
    @abstractmethod
    def on_fault(
        self,
        execution_ctx: ExecutionContext,
        ctx: FaultContext,
        params: dict[str, Any],
    ) -> FaultResponse | None: ...

Returns FaultResponse | None

Independent of every other interface: a notifier may implement this alone, leaving mode() fully SKIP so the policy costs nothing on the normal path. FaultContext does not subclass ResponseContext, but it carries the same field names — ctx.response_status, ctx.response_body and the rest read exactly as they do in on_response_body — and adds the description of the failure:

Field	Meaning
source	Which actor produced the failure — a FaultSource value: gateway (a policy rejection or an engine failure), backend (the upstream's own error status), router (the upstream could not be reached), noRoute, or unknown
fault	The failure's code, class, direction and message, as FaultDetails. None when nothing described it, which includes every backend error
original_status	Status before a policy changed it, 0 when nothing did
policy / policy_version / policy_phase	The policy that caused the failure and the phase it was in. Empty means no policy did — a router failure
response_committed	True when the response already reached the client, so any change here is discarded
route_key	The matched route

A FaultResponse merges over the error the client is receiving: status_code, body, and headers_to_set / headers_to_append / headers_to_remove. Every fault entry runs, in order; use an entry's executionCondition to keep it off failures it should not handle.

Returning None leaves the error exactly as it was, which is what a handler that only notifies or records should do.

---

Processing mode

The mode() method declares which pipeline phases your policy wants to be called for. The gateway uses this to skip unnecessary work.

@dataclass(slots=True)
class ProcessingMode:
    request_header_mode:  HeaderProcessingMode = HeaderProcessingMode.SKIP
    request_body_mode:    BodyProcessingMode   = BodyProcessingMode.SKIP
    response_header_mode: HeaderProcessingMode = HeaderProcessingMode.SKIP
    response_body_mode:   BodyProcessingMode   = BodyProcessingMode.SKIP

HeaderProcessingMode

Value	Meaning
SKIP	Do not invoke the header hook for this policy
PROCESS	Invoke the header hook

BodyProcessingMode

Value	Meaning
SKIP	Do not read the body for this policy
BUFFER	Buffer the entire body and call the body hook once
STREAM	Call the chunk hook for each body segment

---

Actions reference

Return an action dataclass (or None to pass through unchanged) from your policy hook.

Request actions

Dataclass	Effect
UpstreamRequestHeaderModifications	Modify headers / path / method / query params before forwarding to upstream
UpstreamRequestModifications	Modify body and headers / path / method / query params before forwarding
ImmediateResponse	Short-circuit the request and return a response to the client immediately
None	Pass through without modifications

Response actions

Dataclass	Effect
DownstreamResponseHeaderModifications	Modify response headers before sending to client
DownstreamResponseModifications	Modify response body and headers before sending to client
ImmediateResponse	Replace the upstream response with a synthetic one
None	Pass through without modifications

Streaming actions

Dataclass	Effect
ForwardRequestChunk	Forward the (optionally rewritten) chunk upstream
ForwardResponseChunk	Forward the (optionally rewritten) chunk downstream
TerminateResponseChunk	Send a final chunk downstream and end the stream
None	Drop the chunk (use with care)

ImmediateResponse fields

@dataclass(slots=True)
class ImmediateResponse:
    status_code: int = 500
    headers: dict[str, str] = field(default_factory=dict)
    body: bytes | None = None
    analytics_metadata: dict[str, Any] = field(default_factory=dict)
    dynamic_metadata: dict[str, dict[str, Any]] = field(default_factory=dict)
    analytics_header_filter: DropHeaderAction = field(default_factory=DropHeaderAction)
    is_fault: bool = False               # declare a failure the status does not reveal
    fault: FaultDetails | None = None    # describe the failure: code, type, direction, message

A response enters the fault flow when its status is 400 or above or is_fault is True.

---

Context types

ExecutionContext

Injected into every hook. Contains tracing information and a cancellation check.

@dataclass(slots=True)
class ExecutionContext:
    request_id: str
    phase: ExecutionPhase
    deadline: datetime | None = None
    route_name: str = ""
    policy_name: str = ""
    policy_version: str = ""
    trace_id: str | None = None
    span_id: str | None = None

    def is_cancelled(self) -> bool: ...  # check before expensive operations

SharedContext

Accessible via ctx.shared on every context object.

@dataclass(slots=True)
class SharedContext:
    project_id: str
    request_id: str
    api_id: str
    api_name: str
    api_version: str
    api_kind: str
    api_context: str
    operation_path: str
    auth_context: AuthContext | None
    metadata: dict[str, Any]  # cross-policy mutable bag
    resolved_operation: str                # canonical operation; "" unless the kind resolves one
    resolution_attributes: dict[str, str]  # protocol facts from the request; per-request copy

resolved_operation names the operation whose policy chain is running, for an API kind whose operation cannot be read off the route — operation_path is identical for every operation on a multiplexed endpoint, so it cannot tell them apart. resolution_attributes carries what the resolver read out of the request while identifying that operation ("a2a.context.id", "a2a.task.id", …), so your policy does not re-parse the body, and so a request-header-phase policy can see a body-sourced value at all. Both are empty for every API kind that shipped before Agent: read that as "not applicable", not as a failure. Attribute values come out of a request body — treat them as untrusted.

AuthContext

Populated when authentication has run upstream of your policy.

@dataclass(slots=True)
class AuthContext:
    authenticated: bool
    authorized: bool
    auth_type: str      # e.g. "JWT", "APIKey"
    subject: str
    issuer: str
    audience: list[str]
    scopes: dict[str, bool]
    credential_id: str
    properties: dict[str, str]
    previous: AuthContext | None  # previous auth layer result

Headers

Read-only multi-value header wrapper. All lookups are case-insensitive.

headers = ctx.headers

headers.get("content-type")      # -> list[str]
headers.has("authorization")     # -> bool
headers.get_all()                 # -> dict[str, list[str]]

for name, values in headers.iterate():
    print(name, values)

---

Versioned imports

The top-level package always re-exports the latest stable contract. Use these imports in most cases:

from apip_sdk_core import RequestPolicy, ProcessingMode, ImmediateResponse

If you need to pin to a specific contract version (e.g. in a library):

from apip_sdk_core.policy.v1alpha2 import RequestPolicy, ProcessingMode

The current stable contract is v1alpha2.

---

Development

Build

python -m build

Run tests

python -m unittest discover -s tests

Type-check

mypy src/

---

License

Apache License 2.0 — see LICENSE for details.

© 2026 WSO2 LLC.