kindred-ai 0.1.0

Kindred AI SDK - A Python SDK that intercepts HTTP requests and reroutes them based on configuration rules

Kindred AI SDK

A Python SDK that intercepts HTTP requests and reroutes them based on configuration rules. Works with any domain or API provider - not limited to any specific service. Perfect for rerouting LLM calls to caching gateways, mocking payment APIs during development, or centralized logging of third-party API usage.

Features

• 🎯 Intercepts outgoing HTTP requests from requests and httpx libraries
• 🔀 Reroutes requests to custom destinations based on hostname rules
• 🔒 Preserves all headers, query parameters, and request bodies
• 🚀 Supports both sync (requests, httpx.Client) and async (httpx.AsyncClient) clients
• 🎛️ Context manager support for temporary interception
• 📝 Verbose logging option for debugging

Installation

From Local Directory (Development)

cd sidecar/SDK
pip install -e ".[all]"

From PyPI (When Published)

pip install kindred-ai[all]

Quick Setup (3 Steps)

1. Set environment variables:

   # Kindred URL (used to fetch restricted URLs and as the routing destination)
   export KINDRED_URL="http://your-server.com:8080"
   
   # Agent token (also serves as agent ID, provided by Kindred)
   export KINDRED_AGENT_TOKEN="your-token-here"
   

2. Add 2 lines to your code:

   from kindred_ai import Interceptor
   Interceptor.init(verbose=True)
   

3. That's it! HTTP requests whose URLs match your restricted list are now automatically routed through Kindred.

See USER_GUIDE.md for detailed instructions.

Quick Start

Automatic restricted-URL mode (recommended)

Let Kindred manage which URLs are intercepted based on the restricted URLs configured for your agent:

export KINDRED_URL="http://your-server.com:8080"
export KINDRED_AGENT_TOKEN="your-token-here"

from kindred_ai import Interceptor

# Initialize without explicit rules - Kindred_URL + KINDRED_AGENT_TOKEN
# are used to fetch restricted URLs for this agent.
Interceptor.init(verbose=True)

import requests

# Only requests whose URLs are in the restricted list will be intercepted.
response = requests.get("https://api.stripe.com/v1/charges")
print(response.json())

# If there are no restricted URLs, nothing is intercepted.

Basic Usage with Specific Rules

from kindred_ai import Interceptor

# Define routing rules - works with ANY domain/provider
rules = {
    "api.openai.com": "https://my-custom-gateway.com/openai",
    "api.stripe.com": "http://localhost:8080/stripe-mock",
    "api.github.com": "https://proxy.example.com/github"
}

# Initialize the interceptor
Interceptor.init(routing_rules=rules, verbose=True)

# Now all requests matching the rules will be rerouted
import requests
response = requests.get("https://api.stripe.com/v1/charges")
# Actually goes to: http://localhost:8080/stripe-mock/v1/charges

# Stop interception
Interceptor.stop()

Catch-All with Code

You can also specify a catch-all endpoint directly in code:

from kindred_ai import Interceptor

# Route ALL requests to a single endpoint
rules = {
    "*": "http://localhost:8080/proxy"  # "*" matches all hostnames
}

Interceptor.init(routing_rules=rules, verbose=True)
# All API requests now go to localhost:8080/proxy

Using with Context Manager

from kindred_ai import Interceptor

rules = {
    "api.stripe.com": "http://localhost:8080/stripe-mock"
}

# Temporary interception (useful for testing)
with Interceptor.scope(routing_rules=rules, verbose=True):
    import requests
    response = requests.post("https://api.stripe.com/v1/charges", json={...})
    # Request is intercepted and rerouted to localhost:8080

# Outside the context, requests are normal
response = requests.post("https://api.stripe.com/v1/charges", json={...})
# Goes to the original Stripe API

Example: Rerouting LLM API Calls

from kindred_ai import Interceptor
from openai import OpenAI

# Reroute OpenAI API calls to your caching gateway
rules = {
    "api.openai.com": "https://my-cache-gateway.com/openai"
}

Interceptor.init(routing_rules=rules, verbose=True)

client = OpenAI(api_key="sk-...")
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)
# Request is automatically rerouted to your gateway

Interceptor.stop()

Example: Mocking Payment APIs

from kindred_ai import Interceptor

# Reroute Stripe calls to a mock server during development
rules = {
    "api.stripe.com": "http://localhost:8080/stripe-mock"
}

Interceptor.init(routing_rules=rules, verbose=True)
# All Stripe API calls now go to your mock server

API Reference

Interceptor.init(routing_rules=None, agent_token=None, kindred_url=None, verbose=False)

Initialize the interceptor.

Parameters:

• routing_rules (dict, optional): Dictionary mapping original hostnames to new destinations.
  • If provided, these rules are used directly (you can still use "*" / "__all__" for catch‑all behavior).
  • Example: {"api.stripe.com": "http://localhost:8080/stripe-mock", "api.openai.com": "https://my-gateway.com/openai"}
• agent_token (str, optional): Agent token (also serves as agent ID). If not provided, KINDRED_AGENT_TOKEN environment variable is used.
• kindred_url (str, optional): Base Kindred URL. If not provided, KINDRED_URL environment variable is used.
• verbose (bool): Whether to log intercepted requests (default: False)

Behavior when routing_rules is None:

• If agent_token/kindred_url (or the corresponding env vars) are set, the SDK:
  • Fetches restricted URLs from: KINDRED_URL/restricted-urls?agent_id=<KINDRED_AGENT_TOKEN>
  • Builds routing rules from those URLs (one rule per hostname pointing to KINDRED_URL)
  • Initializes interception only if at least one restricted URL is found
• If the fetch fails or no restricted URLs are returned, the interceptor is not initialized (no interception).

Raises:

• ValueError: If routing rules are invalid or neither rules nor the required environment variables are provided

Environment Variables:

• KINDRED_URL: Base Kindred URL, used both to fetch restricted URLs and as the routing destination.
• KINDRED_AGENT_TOKEN: Token used as agent ID when fetching restricted URLs, and automatically added as X-Kindred-Agent-Token header to intercepted requests.

Interceptor.stop()

Stop interception and restore original behavior.

Interceptor.scope(routing_rules, verbose=False)

Context manager for temporary interception.

Parameters:

• routing_rules (dict): Dictionary mapping original hostnames to new destinations
• verbose (bool): Whether to log intercepted requests

Example:

with Interceptor.scope({"api.stripe.com": "http://localhost:8080"}):
    # Requests here are intercepted
    pass
# Requests here are normal

Interceptor.is_active()

Check if the interceptor is currently active.

Returns:

• bool: True if interceptor is initialized, False otherwise

Interceptor.get_rules()

Get the current routing rules.

Returns:

• dict: Copy of the current routing rules dictionary

Routing Rules

Routing rules map original hostnames to new destinations:

rules = {
    "api.openai.com": "https://my-gateway.com/openai",
    "api.stripe.com": "http://localhost:8080/stripe-mock"
}

Catch-All Routing (advanced, manual rules): You can route ALL requests to a single endpoint using a catch‑all key in routing_rules:

rules = {
    "*": "http://localhost:8080/proxy"  # Routes all requests
}
Interceptor.init(routing_rules=rules)

Important Notes:

• The destination must include a hostname (e.g., https://example.com, not /relative/path)
• Paths from the original URL are preserved and appended to the destination path
• Query parameters and fragments are preserved
• Headers (including Authorization) are forwarded to the new destination
• If KINDRED_AGENT_TOKEN environment variable is set, it's automatically added as X-Kindred-Agent-Token header to all intercepted requests
• Original destination URL is added as X-Kindred-Original-URL header
• Original payload/body is added as X-Kindred-Original-Payload header (with type in X-Kindred-Original-Payload-Type)
• Specific hostname rules take precedence over catch-all rules

Example URL Rewriting:

• Original: https://api.stripe.com/v1/charges?limit=10

• Rule: {"api.stripe.com": "http://localhost:8080/stripe-mock"}

• Result: http://localhost:8080/stripe-mock/v1/charges?limit=10

• Original: https://api.openai.com/v1/chat/completions?model=gpt-4

• Rule: {"api.openai.com": "https://my-gateway.com/openai"}

• Result: https://my-gateway.com/openai/v1/chat/completions?model=gpt-4

Supported Libraries

• ✅ requests - Full support for requests.Session.request
• ✅ httpx - Full support for both httpx.Client.request (sync) and httpx.AsyncClient.request (async)

SSL/TLS Considerations

When rerouting traffic from HTTPS to HTTPS, ensure your gateway has valid SSL certificates. If you're using self-signed certificates for local development, you may need to configure the underlying HTTP library to disable SSL verification (not recommended for production):

import requests
import urllib3

# Disable SSL warnings (not recommended for production)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Use verify=False in requests (not recommended for production)
response = requests.get("https://api.example.com/endpoint", verify=False)

Request Headers Added by SDK

The SDK automatically adds several headers to intercepted requests to provide context to your server:

Agent Token Header

If KINDRED_AGENT_TOKEN environment variable is set, it's added as X-Kindred-Agent-Token header. This allows your server to identify which agent/user is making requests.

Setup:

export KINDRED_AGENT_TOKEN="your-token-here"

Original Destination Header

The original destination URL (before rerouting) is always added as X-Kindred-Original-URL header. This tells your server where the request was originally supposed to go.

Example:

• Original request: https://api.stripe.com/v1/charges
• Rerouted to: http://your-server.com:8080/proxy/v1/charges
• Header added: X-Kindred-Original-URL: https://api.stripe.com/v1/charges

Original Payload Headers

The original request payload/body is captured and added as headers:

• X-Kindred-Original-Payload: The payload content
• X-Kindred-Original-Payload-Type: The type (json, string, or base64)

Payload Types:

• JSON: If payload is a dict/list, it's serialized as JSON string
• String: If payload is a string, it's included as-is
• Base64: If payload is bytes, it's base64-encoded

Example:

from kindred_ai import Interceptor
Interceptor.init(verbose=True)

import requests
response = requests.post(
    "https://api.stripe.com/v1/charges",
    json={"amount": 1000, "currency": "usd"}
)
# Headers added:
# X-Kindred-Original-URL: https://api.stripe.com/v1/charges
# X-Kindred-Original-Payload: {"amount": 1000, "currency": "usd"}
# X-Kindred-Original-Payload-Type: json
# X-Kindred-Agent-Token: your-token-here (if set)

Note: Headers are only added to requests that are intercepted (rerouted). Non-intercepted requests are unchanged.

Error Handling

The SDK is designed to fail safely:

• If a library (requests or httpx) is not available, the patcher will skip it with a warning
• If URL parsing fails, the original URL is returned unchanged
• Invalid routing rules raise ValueError during initialization
• If KINDRED_AGENT_TOKEN is not set, requests still work (just without the token header)

Development

Running Tests

# Install test dependencies
pip install -e ".[all]"

# Run tests
python -m pytest tests/
# or
python -m unittest discover tests

Project Structure

kindred_ai/
├── __init__.py           # Public API
├── core.py               # Main interceptor logic
├── rules.py               # URL rewriting logic
├── patchers/
│   ├── __init__.py
│   ├── requests_patch.py # requests library patcher
│   └── httpx_patch.py    # httpx library patcher
└── tests/
    ├── test_requests.py
    └── test_integration.py

License

MIT

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.