harness-sdk (otel-python-sdk)

Generic Python agent SDK with OpenTelemetry instrumentation and a plugin architecture.

Local development

bash scripts/fetch-vendor.sh
pip install -e ".[dev,anthropic,google-genai,openai,litellm]"
./scripts/run-unit-tests.sh

Environment variables for SDK configuration use the HARNESS_ prefix (for example HARNESS_SERVICE_NAME, HARNESS_REPORTING_ENDPOINT). The legacy HA_, AT_, and TA_ prefixes remain supported for backwards compatibility; when the same setting is defined under multiple prefixes, HARNESS_ wins, then HA_, then AT_, then TA_.

Integration tests (MySQL / PostgreSQL)

cd test/externalServices && docker compose up -d --wait
cd ../..
RUN_SDK_INTEGRATION_TESTS=1 ./scripts/run-unit-tests.sh

CI

GitHub Actions workflows (public repo, ubuntu-latest):

Publishing to PyPI

1. Configure trusted publishing on PyPI (and TestPyPI for RCs):
   • PyPI: environment pypi, workflow publish.yaml, job publish-pypi
   • TestPyPI: environment testpypi, workflow publish.yaml, job publish-testpypi
2. Create and push a release tag:

   git tag v0.1.0
   git push origin v0.1.0
   

3. Release candidates (v1.0.0-rc.1) publish to TestPyPI only; stable tags publish to PyPI.

Version is taken from the tag (v1.2.3 → 1.2.3) via scripts/build.sh, which updates src/harness_sdk/version.py and pyproject.toml.

Installing

From PyPI (stable releases)

pip install harness-sdk

Optional instrumentation extras:

pip install "harness-sdk[anthropic,google-genai,openai,litellm]"

From TestPyPI (release candidates)

Tags containing -rc (for example v1.0.0-rc.1) are published only to TestPyPI. Use TestPyPI as the primary index and PyPI as a fallback so dependencies that are not on TestPyPI still resolve:

pip install \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  "harness-sdk==1.0.0-rc.1"

With optional extras:

pip install \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  "harness-sdk[anthropic,google-genai,openai,litellm]==1.0.0-rc.1"

Pin the version to the RC you want (see published files on TestPyPI). The version string matches the tag without the leading v (tag v1.0.0-rc.1 → 1.0.0-rc.1). In a virtualenv or requirements.txt, the same flags apply:

--index-url https://test.pypi.org/simple/
--extra-index-url https://pypi.org/simple/
harness-sdk==1.0.0-rc.1

After install, import and auto-instrument work the same as a PyPI release:

from harness_sdk.agent import Agent

harness-instrument python app.py

Usage

from harness_sdk.agent import Agent

agent = Agent()
agent.instrument()

Auto-instrumentation:

export HARNESS_CONFIG_FILE=/path/to/config.yaml
harness-instrument python app.py

Enabling instrumentation (opt-in)

Instrumentation is opt-in. By default agent.instrument() and harness-instrument instrument nothing; you must explicitly enable the categories you want. A flag is "on" only when its value is (case-insensitively) true; any other value — or an unset variable — leaves it off.

Each AI provider is enabled independently. The HARNESS_ENABLE_* opt-in flags are only read under the HARNESS_ prefix — they have no HA_/AT_/TA_ aliases. The legacy HA_GEN_AI_ENABLED master switch no longer enables or disables AI instrumentation; use the per-provider flags instead.

# Enable HTTP/API + OpenAI only
export HARNESS_ENABLE_API=true
export HARNESS_ENABLE_AI_OPENAI=true
harness-instrument python app.py

agent.instrument(skip_libraries=[...]) still works and takes precedence: a library listed in skip_libraries is never instrumented even if its category is enabled.

Enrich the current span

Add runtime attributes while the target instrumented span is active:

from harness_sdk import set_span_attribute, set_span_attributes

set_span_attribute("request.client.name", client_name)
set_span_attributes({
    "agent.action.type": action_type,
    "custom.retry.count": retry_count,
})

These helpers update only the current recording span. They silently do nothing when no recording span is active. Attributes do not propagate to child spans or downstream services.

Keys may use any customer-defined name. Values must be valid OpenTelemetry attribute values: strings, booleans, integers, floats, or homogeneous sequences of those types. For duplicate keys, the last write wins; instrumentation can overwrite a customer value if it writes the same key later. Because the span already exists, these attributes cannot influence head sampling.

Plugins

The SDK loads extensions via setuptools entry points. Each plugin has a name (the entry-point key). Names are listed in config or environment variables; only installed plugins are loaded, in the order you configure.

Built-in observability plugins (shipped with harness-sdk):

• builtin_pipeline — OTLP export, sampling, exclusion processors
• builtin_span_attributes — service name and configured span attributes

Example agent-config.yaml:

service_name: my-service
reporting:
  endpoint: http://localhost:4318
plugins:
  control:
    - my_policy          # order matters: first plugin runs first
  observability:
    - builtin_pipeline
    - builtin_span_attributes
    - my_exporter        # custom plugin after builtins

Or via environment (comma-separated, same order semantics):

export HA_CONTROL_PLUGINS=my_policy
export HA_OBSERVABILITY_PLUGINS=builtin_pipeline,builtin_span_attributes,my_exporter

Create a control plugin

Control plugins evaluate HTTP/gRPC ingress and GenAI spans. They return a ControlResult (block, headers, span attributes, etc.). Plugins run in config order; the chain stops when one returns block=True.

1. Implement the plugin in your package (see harness_sdk.plugins.control.ControlPlugin):

# my_company_policy/plugin.py
from typing import Any
from opentelemetry.trace import Span
from harness_sdk.plugins.control import ControlResult, ControlPlugin


class MyPolicyPlugin:
    name = "my_policy"
    provides_blocking = True  # set True if this plugin can block requests

    def on_init(self, config: Any) -> None:
        self._config = config

    def evaluate(
        self, span: Span, url: str, headers: dict, body, is_grpc: bool
    ) -> ControlResult:
        result = ControlResult()
        # result.block = True
        # result.response_status_code = 403
        return result

    def evaluate_agent_span(self, span: Span, body: str = "") -> ControlResult:
        return ControlResult()

    def shutdown(self) -> None:
        pass


def factory(config: Any) -> ControlPlugin:
    return MyPolicyPlugin()

2. Register the entry point in your package pyproject.toml:

[project.entry-points.harness_sdk_control_plugin]
my_policy = "my_company_policy.plugin:factory"

(Equivalent in setup.py: entry_points={'harness_sdk_control_plugin': ['my_policy = ...']}.)

3. Install your package in the same environment as the app (pip install my-company-policy).

4. Enable the plugin by name in config or HA_CONTROL_PLUGINS (see above).

For tests or one-off wiring you can also call agent.register_control_plugin(plugin) after Agent() is constructed.

Reference implementation: the Traceable agent ships a control plugin as traceable (traceableai.plugins.traceable_control:factory).

Create an observability plugin

Observability plugins contribute OpenTelemetry SpanProcessor instances to the tracer provider. Processors are registered in config order (each plugin’s create_span_processors may return multiple processors).

1. Implement the plugin (see harness_sdk.plugins.observability.ObservabilityPlugin):

# my_company_telemetry/plugin.py
from typing import Any, List
from opentelemetry.sdk.trace import SpanProcessor
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor


class MyExporterPlugin:
    name = "my_exporter"
    priority = 300  # informational; ordering is driven by config list

    def on_init(self, config: Any) -> None:
        self._config = config

    def create_span_processors(self, config: Any) -> List[SpanProcessor]:
        return [SimpleSpanProcessor(ConsoleSpanExporter())]

    def shutdown(self) -> None:
        pass


def factory(config: Any) -> MyExporterPlugin:
    return MyExporterPlugin()

2. Register the entry point:

[project.entry-points.harness_sdk_observability_plugin]
my_exporter = "my_company_telemetry.plugin:factory"

3. Install your package and enable it under plugins.observability or HA_OBSERVABILITY_PLUGINS.

If you omit observability plugins entirely, the SDK defaults to builtin_pipeline and builtin_span_attributes. Custom plugins typically keep those builtins and append your entry after them.

Reference implementations in this repo:

• harness_sdk.plugins.builtin.pipeline
• harness_sdk.plugins.builtin.span_attributes