Python SDK

The Python SDK ships in the shouldertap package. It offers two integration styles: an HTTP client for a consumer running as its own service, and in-process registration for a consumer embedded in the same process as the engine.

ShoulderTapClient

ShoulderTapClient is a thin HTTP client over the Protocol API. Import it and the contracts you’ll pass:

from shouldertap.sdk.client import ShoulderTapClient
from shouldertap.engine.contracts import ContextRequest

client = ShoulderTapClient(
    base_url="http://localhost:8776/api/v1",   # default
    api_token="...",                            # your SHOULDERTAP_API_TOKEN
)

result = client.ask(
    ContextRequest(
        kind="glossary.definition",
        topic="revenue metrics",
        question="What does 'active customer' mean for Q2 reporting?",
        consumer="bi.assistant",
    )
)
# {"id": "req_01J...", "status": "queued"}

status = client.get_request(result["id"])

Methods

Construct a client. api_token is sent as Authorization: Bearer <token> on every request. Supports use as a context manager (with ShoulderTapClient(...) as client:), which closes the underlying HTTP connection on exit.

Submit a request (POST /requests). Returns { id, status }.

Fetch a request’s current status and, once resolved, its accepted proposal (GET /requests/{id}).

Remove a consumer (DELETE /consumers/{id}).

Close the HTTP connection. Called automatically when used as a context manager.

In-process consumers

If your consumer runs in the same process as the engine, you can skip HTTP and webhooks entirely and receive answers as direct callbacks. Register with register_in_process:

from shouldertap.sdk.client import register_in_process

class Callbacks:
    def on_proposal(self, proposal): ...
    def on_proposal_accepted(self, proposal):
        # write to your system of record here
        ...
    def on_proposal_rejected(self, proposal, reason): ...
    def on_request_failed(self, request_id, reason): ...

register_in_process(deliverer, consumer_id="embedded-consumer", callbacks=Callbacks())

The callback object implements the same four events a webhook consumer receives — on_proposal, on_proposal_accepted, on_proposal_rejected(proposal, reason), and on_request_failed(request_id, reason) — but as in-process method calls with no serialization or network hop.

register_in_process takes a ConsumerDeliverer from shouldertap.engine.delivery, which the engine provides when your consumer is wired into it. Use this style when you’re embedding ShoulderTap as a library; use ShoulderTapClient plus webhooks when your consumer is a separate service.

Get started

Run the engine

Integrate agents

Reference

ShoulderTapClient

Methods

In-process consumers

​ShoulderTapClient

​Methods

​In-process consumers

ShoulderTapClient

Methods

In-process consumers