kindred-ai 0.1.0
Kindred AI SDK - A Python SDK that intercepts HTTP requests and reroutes them based on configuration rules
License: MIT
Requires Python: >=3.8

# 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)

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

### From PyPI (When Published)

```bash
pip install kindred-ai[all]
```

## Quick Setup (3 Steps)

1. **Set environment variables:**
   ```bash
   # 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:**
   ```python
   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](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:

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

```python
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

```python
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:

```python
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

```python
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

```python
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

```python
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:**
```python
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:

```python
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`:
   ```python
   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):

```python
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:**
```bash
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:**
```python
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

```bash
# 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.