OpaClient - Open Policy Agent Python Client

OpaClient is a Python client library designed to interact with the Open Policy Agent (OPA). It supports both synchronous and asynchronous requests, making it easy to manage policies, data, and evaluate rules in OPA servers.

Features

Manage Policies: Create, update, retrieve, and delete policies.
Manage Data: Create, update, retrieve, and delete data in OPA.
Evaluate Policies: Use input data to evaluate policies and return decisions.
Synchronous & Asynchronous: Choose between sync or async operations to suit your application.
SSL/TLS Support: Communicate securely with SSL/TLS, including client certificates.
Customizable: Use custom headers, timeouts, and other configurations.

Installation

You can install the OpaClient package via pip:

pip install opa-python-client

Quick Start

Synchronous Client Example

from opa_client.opa import OpaClient

# Initialize the OPA client
client = OpaClient(host='localhost', port=8181)

# Check the OPA server connection
try:
    print(client.check_connection())  # True
finally:
    client.close_connection()

or with client factory

from opa_client import create_opa_client

client = create_opa_client(host="localhost", port=8181)

Check OPA healthy. If you want check bundels or plugins, add query params for this.

from opa_client.opa import OpaClient

client = OpaClient()

print(client.check_health()) # response is  True or False
print(client.check_health({"bundle": True})) # response is  True or False
# If your diagnostic url different than default url, you can provide it.
print(client.check_health(diagnostic_url="http://localhost:8282/health"))  # response is  True or False
print(client.check_health(query={"bundle": True}, diagnostic_url="http://localhost:8282/health"))  # response is  True or False

Asynchronous Client Example

import asyncio
from opa_client.opa_async import AsyncOpaClient

async def main():
    async with AsyncOpaClient(host='localhost', port=8181) as client:
        result = await client.check_connection()
        print(result)

# Run the async main function
asyncio.run(main())

or with clien factory

from opa_client import create_opa_client

client = create_opa_client(async_mode=True,host="localhost", port=8181)

Authentication with a Bearer Token

If your OPA server is configured with token-based authentication, pass a token and the client will send it as an Authorization: Bearer <token> header on every request:

from opa_client.opa import OpaClient

client = OpaClient(host='opa.example.com', token='my-secret-token')

This works the same way for AsyncOpaClient.

Waiting for OPA to Become Ready

Useful during container/CI startup, when OPA may not be immediately reachable:

from opa_client.opa import OpaClient

client = OpaClient()
client.wait_for_ready(timeout=10)  # blocks until healthy, or raises ConnectionsError

Asynchronous:

await client.wait_for_ready(timeout=10)

Secure Connection with SSL/TLS

You can use OpaClient with secure SSL/TLS connections, including mutual TLS (mTLS), by providing a client certificate and key.

Synchronous Client with SSL/TLS

from opa_client.opa import OpaClient

# Path to your certificate and private key
cert_path = '/path/to/client_cert.pem'
key_path = '/path/to/client_key.pem'

# Initialize the OPA client with SSL/TLS
client = OpaClient(
    host='your-opa-server.com',
    port=443,  # Typically for HTTPS
    ssl=True,
    cert=(cert_path, key_path)  # Provide the certificate and key as a tuple
)

# Check the OPA server connection
try:
    result = client.check_connection()
    print(result)
finally:
    client.close_connection()

Asynchronous Client with SSL/TLS

import asyncio
from opa_client.opa_async import AsyncOpaClient

# Path to your certificate and private key
cert_path = '/path/to/client_cert.pem'
key_path = '/path/to/client_key.pem'

async def main():
    # Initialize the OPA client with SSL/TLS
    async with AsyncOpaClient(
        host='your-opa-server.com',
        port=443,  # Typically for HTTPS
        ssl=True,
        cert=(cert_path, key_path)  # Provide the certificate and key as a tuple
    ) as client:
        # Check the OPA server connection
        result = await client.check_connection()
        print(result)

# Run the async main function
asyncio.run(main())

Usage

Policy Management

Create or Update a Policy

You can create or update a policy using the following syntax:

Synchronous:

policy_name = 'example_policy'
policy_content = '''
package example

default allow = false

allow if {
    input.user == "admin"
}
'''

client.update_policy_from_string(policy_content, policy_name)

Asynchronous:

await client.update_policy_from_string(policy_content, policy_name)

OPA 1.0+ compatibility: Policies should use Rego v1 syntax (allow if { ... }). The client accepts legacy v0 policies as well — when uploading to OPA 1.0+, it automatically upgrades common v0 constructs and retries. Disable this with rego_compat=False on update_policy_from_string.

Or from url:

Synchronous:

policy_name = 'example_policy'

client.update_policy_from_url("http://opapolicyurlexample.test/example.rego", policy_name)

Asynchronous:

await client.update_policy_from_url("http://opapolicyurlexample.test/example.rego", policy_name)

Update policy from rego file

client.update_opa_policy_fromfile("/your/path/filename.rego", endpoint="fromfile") # response is True

client.get_policies_list()

Asynchronous:

await client.update_opa_policy_fromfile("/your/path/filename.rego", endpoint="fromfile") # response is True

await client.get_policies_list()

Retrieve a Policy

After creating a policy, you can retrieve it:

Synchronous:

policy = client.get_policy('example_policy')
print(policy)
# or
policies = client.get_policies_list()
print(policies)

Asynchronous:

policy = await client.get_policy('example_policy')
print(policy)

# or
policies = await client.get_policies_list()
print(policies)

Save policy to file from OPA service

client.policy_to_file(policy_name="example_policy",path="/your/path",filename="example.rego")

Asynchronous:

await client.policy_to_file(policy_name="example_policy",path="/your/path",filename="example.rego")

Information about policy path and rules

print(client.get_policies_info())
#{'example_policy': {'path': 'http://localhost:8181/v1/data/example', 'rules': ['http://localhost:8181/v1/data/example/allow']}}

Asynchronous:

print(await client.get_policies_info())
#{'example_policy': {'path': 'http://localhost:8181/v1/data/example', 'rules': ['http://localhost:8181/v1/data/example/allow']}}

Delete a Policy

You can delete a policy by name:

Synchronous:

client.delete_policy('example_policy')

Asynchronous:

await client.delete_policy('example_policy')

Data Management

Create or Update Data

You can upload arbitrary data to OPA:

Synchronous:

data_name = 'users'
data_content = {
    "users": [
        {"name": "alice", "role": "admin"},
        {"name": "bob", "role": "user"}
    ]
}

client.update_or_create_data(data_content, data_name)

Asynchronous:

await client.update_or_create_data(data_content, data_name)

Retrieve Data

You can fetch the data stored in OPA:

Synchronous:

data = client.get_data('users')
print(data)
# You can use query params for additional info
# provenance - If parameter is true, response will include build/version info in addition to the result.
# metrics - Return query performance metrics in addition to result 
data = client.get_data('users',query_params={"provenance": True})
print(data) # {'provenance': {'version': '0.68.0', 'build_commit': 'db53d77c482676fadd53bc67a10cf75b3d0ce00b', 'build_timestamp': '2024-08-29T15:23:19Z', 'build_hostname': '3aae2b82a15f'}, 'result': {'users': [{'name': 'alice', 'role': 'admin'}, {'name': 'bob', 'role': 'user'}]}}


data = client.get_data('users',query_params={"metrics": True})
print(data) # {'metrics': {'counter_server_query_cache_hit': 0, 'timer_rego_external_resolve_ns': 7875, 'timer_rego_input_parse_ns': 875, 'timer_rego_query_compile_ns': 501083, 'timer_rego_query_eval_ns': 50250, 'timer_rego_query_parse_ns': 199917, 'timer_server_handler_ns': 1031291}, 'result': {'users': [{'name': 'alice', 'role': 'admin'}, {'name': 'bob', 'role': 'user'}]}}

Asynchronous:

data = await client.get_data('users')
print(data)

Delete Data

To delete data from OPA:

Synchronous:

client.delete_data('users')

Asynchronous:

await client.delete_data('users')

Partially Update Data (JSON Patch)

You can partially update data already stored in OPA using a JSON Patch (RFC 6902) document, instead of replacing the whole document with update_or_create_data:

Synchronous:

client.update_or_create_data({"users": {"alice": {"role": "admin"}}}, "acl")

client.patch_data("acl", [
    {"op": "add", "path": "/users/bob", "value": {"role": "user"}},
])
print(client.get_data("acl"))
# {'result': {'users': {'alice': {'role': 'admin'}, 'bob': {'role': 'user'}}}}

Asynchronous:

await client.update_or_create_data({"users": {"alice": {"role": "admin"}}}, "acl")

await client.patch_data("acl", [
    {"op": "add", "path": "/users/bob", "value": {"role": "user"}},
])
print(await client.get_data("acl"))

Server Configuration & Metrics

Get Server Configuration

Retrieve OPA's active configuration (e.g. to check labels, decision logging, or bundle settings):

Synchronous:

print(client.get_config())
# {'result': {'default_decision': '/system/main', 'labels': {'id': '...', 'version': '0.68.0'}}}

Asynchronous:

print(await client.get_config())

Get Server Metrics

Retrieve Prometheus-formatted performance metrics from OPA:

Synchronous:

print(client.get_metrics())
# '# HELP go_info Information about the Go environment.\n# TYPE go_info gauge\ngo_info{version="go1.23.0"} 1\n...'

Asynchronous:

print(await client.get_metrics())

Get Server Status

Retrieve OPA's bundle activation, discovery, and plugin status (requires the status plugin to be enabled on the server):

Synchronous:

print(client.get_status())
# {'result': {'labels': {...}, 'bundles': {...}, 'plugins': {...}}}

Asynchronous:

print(await client.get_status())

If the status plugin isn't enabled, OPA itself responds with a 500 and a "status plugin not enabled" message. get_status() doesn't hide this behind a generic connection error — it raises a normal requests.exceptions.HTTPError (sync) or aiohttp.ClientResponseError (async), with OPA's real response still accessible (e.g. e.response.json()["message"] on the sync client).

Policy Evaluation

Check Permission (Policy Evaluation)

Evaluate a rule from a known package path. This is the recommended method for evaluating OPA decisions.

rego = """
package play

default hello = false

hello {
    m := input.message
    m == "world"
}
"""

check_data = {"message": "world"}

client.update_policy_from_string(rego, "test")
print(client.query_rule(input_data=check_data, package_path="play", rule_name="hello")) # {'result': True}

Asynchronous:

rego = """
package play

default hello = false

hello {
    m := input.message
    m == "world"
}
"""

check_data = {"message": "world"}

await client.update_policy_from_string(rego, "test")
print(await client.query_rule(input_data=check_data, package_path="play", rule_name="hello")) # {'result': True}

You can pass OPA query parameters (e.g. explain, metrics, pretty, instrument) via query_params:

print(client.query_rule(
    input_data=check_data,
    package_path="play",
    rule_name="hello",
    query_params={"metrics": True},
))

Ad-hoc Queries

Execute ad-hoc queries directly:

Synchronous:

data = {
    "user_roles": {
        "alice": [
            "admin"
        ],
        "bob": [
            "employee",
            "billing"
        ],
        "eve": [
            "customer"
        ]
    }
}
input_data = {"user": "admin"}
client.update_or_create_data(data, "userinfo")

result = client.ad_hoc_query(query="data.userinfo.user_roles[name]")
print(result) # {'result': [{'name': 'alice'}, {'name': 'bob'}, {'name': 'eve'}]}

Asynchronous:

data = {
    "user_roles": {
        "alice": [
            "admin"
        ],
        "bob": [
            "employee",
            "billing"
        ],
        "eve": [
            "customer"
        ]
    }
}
input_data = {"user": "admin"}
await client.update_or_create_data(data, "userinfo")

result = await client.ad_hoc_query(query="data.userinfo.user_roles[name]")
print(result) # {'result': [{'name': 'alice'}, {'name': 'bob'}, {'name': 'eve'}]}

ad_hoc_query also accepts query_params for explain, metrics, pretty, etc., the same way query_rule does.

Bulk Rule Evaluation

Note: Open-source OPA has no dedicated "batch decision" REST endpoint — that only exists in Styra's discontinued Enterprise OPA fork. bulk_query_rule gets you the same practical benefit (evaluating one rule against many inputs in a single HTTP round-trip) by building one ad-hoc query that overrides input per item with Rego's with keyword, which works against any standard OPA server.

Use bulk_query_rule when you need a policy decision for many resources/users at once (e.g. filtering a list by permission) without making one query_rule HTTP call per item:

Synchronous:

rego = """
package app.abac

default allow = false

allow {
    input.role == "admin"
}
"""
client.update_policy_from_string(rego, "abac")

# Pass a list for positional results:
result = client.bulk_query_rule(
    [{"role": "admin"}, {"role": "user"}],
    package_path="app.abac",
    rule_name="allow",
)
print(result)  # [True, False]

# Or a dict to get results keyed by your own IDs:
result = client.bulk_query_rule(
    {"alice": {"role": "admin"}, "bob": {"role": "user"}},
    package_path="app.abac",
    rule_name="allow",
)
print(result)  # {'alice': True, 'bob': False}

Asynchronous:

result = await client.bulk_query_rule(
    [{"role": "admin"}, {"role": "user"}],
    package_path="app.abac",
    rule_name="allow",
)
print(result)  # [True, False]

Notes:

package_path and rule_name must be valid Rego identifiers (dot-separated for package_path); a ValueError is raised otherwise.
The rule must resolve to a defined value for every input (e.g. via a default declaration). If it's undefined for any input, bulk_query_rule raises QueryExecuteError rather than silently dropping that result.
An empty inputs list/dict short-circuits to []/{} without making a request.

Compile API (Partial Evaluation)

Use compile_query to partially evaluate a query against a set of unknowns. If the query fully resolves given input_data, the result is unconditionally true/false; otherwise OPA returns a residual set of queries (e.g. usable as a filter for a downstream data store).

Synchronous:

policy = """
package authz

default allow = false

allow if {
    input.user.role == "admin"
}
"""
client.update_policy_from_string(policy, "authz")

# Fully resolved: input is fully known, so the query reduces to true.
result = client.compile_query(
    "data.authz.allow == true",
    input_data={"user": {"role": "admin"}},
    unknowns=[],
)
print(result) # {'result': {'queries': [[]]}}

# Partial evaluation: leave input.user.role unknown to get a residual query.
partial = client.compile_query(
    "data.authz.allow == true",
    unknowns=["input.user.role"],
)
print(partial) # {'result': {'queries': [[{'index': 0, 'terms': [...]}]]}}

Asynchronous:

await client.update_policy_from_string(policy, "authz")

result = await client.compile_query(
    "data.authz.allow == true",
    input_data={"user": {"role": "admin"}},
    unknowns=[],
)
print(result) # {'result': {'queries': [[]]}}

API Reference

Synchronous Client (OpaClient)

check_connection(): Verify connection to OPA server.
get_policies_list(): Get a list of all policies.
get_policies_info(): Returns information about each policy, including policy path and policy rules.
get_policy(policy_name): Fetch a specific policy.
policy_to_file(policy_name): Save an OPA policy to a file..
update_policy_from_string(policy_content, policy_name): Upload or update a policy using its string content.
update_policy_from_url(url,endpoint): Update OPA policy by fetching it from a URL.
update_policy_from_file(filepath,endpoint): Update OPA policy using a policy file.
delete_policy(policy_name): Delete a specific policy.
update_or_create_data(data_content, data_name): Create or update data in OPA.
get_data(data_name): Retrieve data from OPA.
patch_data(data_name, patches): Partially update data using a JSON Patch (RFC 6902) document.
delete_data(data_name): Delete data from OPA.
query_rule(input_data, package_path, rule_name, query_params): Query a specific rule in a package.
ad_hoc_query(query, input_data, query_params): Run an ad-hoc query.
bulk_query_rule(inputs, package_path, rule_name, query_params): Evaluate the same rule against many inputs (list or dict) in a single request.
compile_query(query, input_data, unknowns, options): Partially evaluate a query using the Compile API.
get_config(): Get OPA's active server configuration.
get_metrics(): Get Prometheus-formatted server performance metrics.
get_status(): Get OPA's bundle/discovery/plugin status.
wait_for_ready(timeout, interval): Block until OPA reports healthy, or raise ConnectionsError.

Asynchronous Client (AsyncOpaClient)

Same as the synchronous client, but all methods are asynchronous and must be awaited.

Command-Line Interface

Installing the package also installs an opa-client CLI for scripting and CI use:

opa-client --host localhost --port 8181 health
opa-client wait-for-ready --max-wait 10
opa-client list-policies
opa-client put-policy example --file ./example.rego
opa-client get-policy example
opa-client put-data acl --file ./acl.json
opa-client get-data acl
opa-client query-rule play hello --input '{"message": "world"}'
opa-client bulk-query-rule app.abac allow --inputs-json '[{"role": "admin"}, {"role": "user"}]'
opa-client query "data.play.hello == true" --input '{"message": "world"}'
opa-client status
opa-client config
opa-client metrics

Use --ssl and --token <token> for secured servers. Run opa-client <command> --help for the full option list of any subcommand.

Contributing

Contributions are welcome! Feel free to open issues, fork the repo, and submit pull requests.

License

This project is licensed under the MIT License.

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
opa_client		opa_client
.gitignore		.gitignore
CONTRIBUTING.md		CONTRIBUTING.md
LICENCE.md		LICENCE.md
README.md		README.md
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt
uv.lock		uv.lock

Latest commit

History

Folders and files

Repository files navigation

OpaClient - Open Policy Agent Python Client

Features

Installation

Quick Start

Synchronous Client Example

Asynchronous Client Example

Authentication with a Bearer Token

Waiting for OPA to Become Ready

Secure Connection with SSL/TLS

Synchronous Client with SSL/TLS

Asynchronous Client with SSL/TLS

Usage

Policy Management

Create or Update a Policy

Retrieve a Policy

Delete a Policy

Data Management

Create or Update Data

Retrieve Data

Delete Data

Partially Update Data (JSON Patch)

Server Configuration & Metrics

Get Server Configuration

Get Server Metrics

Get Server Status

Policy Evaluation

Check Permission (Policy Evaluation)

Ad-hoc Queries

Bulk Rule Evaluation

Compile API (Partial Evaluation)

API Reference

Synchronous Client (OpaClient)

Asynchronous Client (AsyncOpaClient)

Command-Line Interface

Contributing

License

About

Topics

Resources

Contributing

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages