Python SDK

uv venv --python 3.13
uv add /path/to/communitai/packages/sdk-python

from communitai_moderation import CommunitAIModerator
import os

# Initialize the client
client = CommunitAIModerator(api_key=os.environ["COMMUNITAI_API_KEY"])

# Moderate content
result = client.moderate(
    content="Hello, world!",
    content_type="text",
)

print(result.decision)    # "approve" | "reject" | "warn" | "review"
print(result.confidence)

client = CommunitAIModerator(
    api_key=os.environ["COMMUNITAI_API_KEY"],
    timeout=10.0,
    max_retries=5,
    debug=True,
)

# Synchronous (blocking)
from communitai_moderation import CommunitAIModerator

client = CommunitAIModerator(api_key="...")

with client:
    result = client.moderate(content="Hello!")


# Asynchronous (non-blocking)
from communitai_moderation import AsyncCommunitAIModerator
import asyncio

async def main():
    async with AsyncCommunitAIModerator(api_key="...") as client:
        result = await client.moderate(content="Hello!")
        print(result.decision)

asyncio.run(main())

from communitai_moderation import (
    ContentType,
    ModerationContext,
    ModerationOptions,
    RulesConfig,
    RuleInterpretation,
)

# Basic moderation
result = client.moderate(
    content="Check out my post!",
    content_type=ContentType.TEXT,
)

# With context and options
result = client.moderate(
    content="Buy now! Limited time offer!",
    content_type=ContentType.TEXT,
    context=ModerationContext(
        community_id="marketplace",
        author_reputation=75,
        metadata={
            "post_type": "listing",
            "category": "electronics",
        },
    ),
    rules_config=RulesConfig(
        use_global_safety=True,
        interpretation=RuleInterpretation.MODERATE,
    ),
    options=ModerationOptions(
        explain=True,
    ),
)

# Handle the result
match result.decision:
    case "approve":
        publish_post(post)
    case "reject":
        print(f"Rejected: {result.explanation}")
    case "warn":
        publish_with_flag(post)
    case "review":
        queue_for_review(post, result)

from communitai_moderation import BatchRequest, BatchItem, Priority

# Submit a batch
batch = client.moderate_batch(
    BatchRequest(
        items=[
            BatchItem(external_id=post.id, content=post.body)
            for post in posts
        ],
        priority=Priority.NORMAL,
    )
)

print(f"Submitted batch {batch.job_id}")

# Poll for status
import time

status = client.get_batch_status(batch.job_id)

while status.status in ("queued", "processing"):
    print(f"Progress: {status.items_processed}/{status.items_total}")
    time.sleep(1)
    status = client.get_batch_status(batch.job_id)

# Get results with data
status = client.get_batch_status(batch.job_id, include_results=True)

for result in status.results or []:
    print(f"{result.external_id}: {result.decision}")

from communitai_moderation import (
    CreateRuleRequest,
    RuleDefinition,
    RuleType,
    RuleAction,
    RuleSeverity,
)

# Create a rule
rule = client.create_rule(
    CreateRuleRequest(
        name="No Spam Links",
        description="Block messages with suspicious links",
        rule_type=RuleType.COMBINED,
        rule_definition=RuleDefinition(
            keywords=["click here"],
            patterns=[r"bit\.ly/\w+"],
            semantic_categories=["spam"],
            semantic_threshold=0.7,
        ),
        action=RuleAction.REJECT,
        severity=RuleSeverity.MODERATE,
    )
)

# List rules
rules = client.list_rules()

for r in rules:
    print(f"{r.name}: {r.action.value}")

# Update a rule
from communitai_moderation import UpdateRuleRequest

updated = client.update_rule(
    rule.id,
    UpdateRuleRequest(is_active=False),
)

# Delete a rule
client.delete_rule(rule.id)

# Get analytics
analytics = client.get_analytics(
    days=30,
    include=["decisions", "latency", "usage"],
)

if analytics.decisions:
    print(f"Total requests: {analytics.decisions.total_requests}")
    print(f"Approvals: {analytics.decisions.decisions.approve}")
    print(f"Rejections: {analytics.decisions.decisions.reject}")

if analytics.latency:
    print(f"P50 latency: {analytics.latency.p50_ms}ms")
    print(f"P99 latency: {analytics.latency.p99_ms}ms")

if analytics.usage:
    current = analytics.usage.current_month
    print(f"Requests this month: {current.total_requests}")
    print(f"Projected month-end: {current.projected_month_end}")

from communitai_moderation.errors import (
    CommunitAIError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    QuotaExceededError,
    ServerError,
    RetryExhaustedError,
    TimeoutError,
)

try:
    result = client.moderate(content="...")
except AuthenticationError as e:
    # Invalid or missing API key
    print(f"Auth failed: {e.message}")
except RateLimitError as e:
    # Too many requests
    print(f"Rate limited: {e.message}")
except QuotaExceededError as e:
    # Monthly quota exceeded
    print("Quota exceeded. Upgrade your plan.")
except ValidationError as e:
    # Invalid request parameters
    print(f"Validation failed: {e.details}")
except ServerError as e:
    # Server-side error
    print(f"Server error: {e.message}")
except RetryExhaustedError as e:
    # All retries failed
    print(f"Failed after {e.attempts} attempts: {e.last_error}")
except TimeoutError as e:
    # Request timed out
    print(f"Request timed out")
except CommunitAIError as e:
    # Other API error
    print(f"API error: {e.code} - {e.message}")

# Sync client
with CommunitAIModerator(api_key="...") as client:
    result = client.moderate(content="Hello!")
# Connection is automatically closed

# Async client
async with AsyncCommunitAIModerator(api_key="...") as client:
    result = await client.moderate(content="Hello!")
# Connection is automatically closed

# Manual cleanup (if not using context manager)
client = CommunitAIModerator(api_key="...")
try:
    result = client.moderate(content="Hello!")
finally:
    client.close()

from communitai_moderation import (
    # Configuration
    CommunitAIConfig,

    # Moderation
    ContentType,
    ModerationContext,
    ModerationOptions,
    RulesConfig,
    RuleInterpretation,
    ModerationDecision,
    ModerationResult,
    RuleTriggered,

    # Batch
    BatchItem,
    BatchRequest,
    BatchResponse,
    BatchStatus,
    BatchStatusResponse,
    BatchItemResult,
    Priority,

    # Rules
    CustomRule,
    CreateRuleRequest,
    UpdateRuleRequest,
    RuleDefinition,
    RuleType,
    RuleAction,
    RuleSeverity,

    # Analytics
    AnalyticsResponse,
    DecisionAnalytics,
    LatencyAnalytics,
    UsageAnalytics,
    CustomRulesAnalytics,
)

# Type checking works with Pydantic models
def process_result(result: ModerationResult) -> None:
    if result.decision == ModerationDecision.REJECT:
        print(f"Rejected with confidence {result.confidence}")