reo-census-mcp 0.2.0

Product usage HTTP client for Reo (POST JSON to ingest.reo.dev/api/product/usage)

reo-census-mcp (Python)

This README is for teams integrating Reo into customer-facing products: you ship applications or services your customers rely on, and usage data is sent to Reo telemetry from that software.

The SDK is stdlib-only, non-blocking by default, so customer-facing request paths stay responsive while events reach Reo telemetry.

Install

pip install reo-census-mcp

ReoProductUsageLogger

Two examples below:

• Version 1 — required payload fields only; set REO_API_KEY in the environment without passing api_key=.
• Version 2 — every keyword, including api_key= in code for the full example.

Version 1 — required payload fields only

Merged payload must include activity_type, user_id, user_id_type, and product_id. source and environment default to PRODUCT_CLOUD and PRODUCTION when you omit them (override per call, on the constructor, or with REO_PRODUCT_USAGE_SOURCE / REO_PRODUCT_USAGE_ENVIRONMENT).

Set REO_API_KEY before constructing the logger (do not pass api_key here—it is read from env).

from reo_census_mcp import ReoProductUsageLogger

logger = ReoProductUsageLogger(
    activity_type="LOGIN_ACTIVITY",
    user_id="https://www.linkedin.com/in/userid",
    user_id_type="LINKEDIN",
    product_id="reoWebApp",
)
ok = logger.log_usage()  # non-blocking by default

Version 2 — all constructor parameters

Every supported keyword, including api_key= in code (Version 1 uses REO_API_KEY env only). event_id / event_at below illustrate the full surface area—omit them in real traffic so each log_usage() auto-fills.

from reo_census_mcp import ReoProductUsageLogger

logger = ReoProductUsageLogger(
    api_key="YOUR_API_KEY",
    endpoint_url="https://ingest.reo.dev/api/product/usage",
    timeout=3.0,
    blocking=True,
    activity_type="LOGIN_ACTIVITY",
    source="PRODUCT_CLOUD",
    environment="PRODUCTION",
    user_id="https://www.linkedin.com/in/userid",
    user_id_type="LINKEDIN",
    ip_addr="156.59.87.83",
    product_id="reoWebApp",
    user_agent="Mozilla/5.0 ...",
    meta={"property1": "value1", "property2": "value2"},
    event_id=1231231232,
    event_at=639303296,
)
ok = logger.log_usage()

Payload fields

The SDK does not reject incomplete payloads locally; ingestion may enforce its own rules. When instrumenting customer-facing products, make sure each merged payload is complete and aligned with your privacy commitments (what you disclose to your customers’ end users, and fields like user_id / meta).

For a valid integration event, supply every required field via the ReoProductUsageLogger constructor, via log_usage, or split across both (constructor merged first — log_usage overrides overlapping keys.)

Payload key	Requirement	Notes
activity_type	Required	e.g. LOGIN_ACTIVITY
source	Optional	Defaults to PRODUCT_CLOUD; override with ctor / log_usage / REO_PRODUCT_USAGE_SOURCE
environment	Optional	Defaults to PRODUCTION; override with ctor / log_usage / REO_PRODUCT_USAGE_ENVIRONMENT
user_id	Required	e.g. LinkedIn URL or stable id
user_id_type	Required	e.g. LINKEDIN
product_id	Required	e.g. reoWebApp
ip_addr	Optional	Omit on the customer-facing client if unknown; server-side code may fill when available
meta	Optional	Arbitrary JSON object
event_id	Auto if omitted everywhere	Omit on constructor and calls for a new id per send; only set once if you intentionally freeze it
event_at	Auto if omitted everywhere	Same as event_id
user_agent	Auto if omitted	SDK default or REO_PRODUCT_USAGE_USER_AGENT

Usage

Your customer-facing Python service (web app, worker, agent, etc.) can mirror this contract. Equivalent curl:

curl --location 'https://ingest.reo.dev/api/product/usage' \
  --header 'X-API-KEY: <API_KEY>' \
  --header 'Content-Type: application/json' \
  --data '{"payload":{...}}'

Python: minimal (required fields only)

from reo_census_mcp import ReoProductUsageLogger

# REO_API_KEY in environment — omit api_key=
logger = ReoProductUsageLogger(
    timeout=3.0,
    activity_type="LOGIN_ACTIVITY",
    source="PRODUCT_CLOUD",
    environment="PRODUCTION",
    product_id="reoWebApp",
)

ok = logger.log_usage(
    user_id="https://www.linkedin.com/in/userid",
    user_id_type="LINKEDIN",
)
# Omit blocking → False (non-blocking): default for customer-facing products and long-running apps.

Python: full payload entirely on constructor (optional)

Use when the merged payload is fixed for this logger (REO_API_KEY env; no api_key=):

logger = ReoProductUsageLogger(
    timeout=3.0,
    activity_type="LOGIN_ACTIVITY",
    source="PRODUCT_CLOUD",
    environment="PRODUCTION",
    user_id="https://www.linkedin.com/in/userid",
    user_id_type="LINKEDIN",
    ip_addr="156.59.87.83",
    product_id="reoWebApp",
    user_agent="Mozilla/5.0 ...",
    meta={"property1": "value1"},
)
logger.log_usage()  # event_id / event_at unset → auto-filled each send

Python: full payload (mix constructor + overrides on log_usage)

Assume you already constructed logger = ReoProductUsageLogger(...) with steady product defaults:

Recommended for customer-facing products: blocking=False (omit it — that is the default). The POST runs on a daemon thread so you do not add latency on user-facing paths.

logger.log_usage(
    activity_type="LOGIN_ACTIVITY",
    source="PRODUCT_CLOUD",
    environment="PRODUCTION",
    user_id="https://www.linkedin.com/in/userid",
    user_id_type="LINKEDIN",
    ip_addr="156.59.87.83",
    event_id=1231231232,
    event_at=639303296,
    product_id="reoWebApp",
    user_agent="Mozilla/5.0 ...",
    meta={"property1": "value1", "property2": "value2"},
)

Synchronous sends (blocking=True)

Only when the process exits right away (CLI, Cron, Lambda-style short workers, tests) — otherwise it may terminate before the background thread completes the request.

logger = ReoProductUsageLogger(..., blocking=True)
logger.log_usage()
# optional: synchronous one-shot on otherwise non-blocking instances
logger.log_usage(..., blocking=True)

• blocking on logger: Constructor sets the instance default (blocking=False for customer-facing throughput). Omit blocking= on log_usage to use it; pass blocking=True on log_usage only for one-off synchronous sends.

• blocking=False: True if queued, False if opted out, invalid URL, missing API key, or body too large (not proof of HTTP 2xx).

• blocking=True: True only after HTTP 2xx (with retries).

If you omit event_id / event_at, they are filled automatically (event_id from a time-based integer, event_at as int(time.time())). If you omit user_agent on the call, the SDK sets a default (reo-census-mcp/<version> or REO_PRODUCT_USAGE_USER_AGENT).

API key

REO_API_KEY must be set in your deployment environment or secret manager, or you must pass a non-empty api_key= when constructing ReoProductUsageLogger. Empty or whitespace-only values are treated as missing.

Opt-out

Uses the same variables as reo-census so your customers (or admins running your customer-facing software) can turn off sending — document these in your privacy or deployment guide:

• PACKAGE_TRACKER_ANALYTICS=false
• DO_NOT_TRACK set to 1, true, or yes (case-insensitive)

Endpoint override

export REO_CENSUS_MCP_ENDPOINT="https://your-host/api/product/usage"

Only http:// and https:// URLs with a host are accepted.

Verbose logging

export PACKAGE_TRACKER_VERBOSE=true

Prints send details to stderr.