admesh-weave-python 0.1.0

AdMesh Backend SDK for Python - Fetch and weave recommendations into LLM responses

admesh-weave-python

Lightweight backend SDK for Python that fetches recommendations from the AdMesh Protocol service. This SDK enables AI platforms to integrate AdMesh's Weave Ad Format into their LLM responses.

Overview

The admesh-weave-python SDK is a thin client wrapper that:

• Fetches recommendations directly from admesh-protocol's /agent/recommend endpoint via HTTP POST
• Uses database-backed caching for instant retrieval of previously generated recommendations
• Formats recommendations for seamless integration into LLM prompts
• Falls back gracefully if no recommendations are available

Architecture

Your Application
    ↓ (calls SDK)
admesh-weave-python SDK
    ↓ (HTTP POST)
admesh-protocol /agent/recommend
    ↓ (checks database cache)
Database Cache Hit? → Return Cached Recommendations
    ↓ (cache miss)
Generate New Recommendations → Save to Database → Return

The SDK uses a simplified database-backed architecture:

• No Pub/Sub: Direct HTTP calls to /agent/recommend endpoint
• No SSE: Synchronous request/response pattern
• Database caching: Recommendations are cached in Firestore with 60-second TTL
• Fast cache hits: Second call returns instantly from database (< 100ms)
• Simple integration: Single HTTP POST, no complex subscription management

Installation

pip install admesh-weave-python

Quick Start

Async Usage (Recommended)

from admesh_weave import AdMeshClient

# Initialize the SDK with your API key
client = AdMeshClient(api_key="your-api-key")

# Get recommendations for Weave
result = await client.get_recommendations_for_weave(
    session_id="sess_123",
    message_id="msg_1",
    query="best laptops for programming",
    timeout_ms=30000
)

if result["found"]:
    print("Recommendations:", result["recommendations"])
    print("Query:", result["query"])
else:
    print("No recommendations found:", result.get("error"))

Synchronous Usage

from admesh_weave import AdMeshClient

client = AdMeshClient(api_key="your-api-key")

# Synchronous version
result = client.get_recommendations_for_weave_sync(
    session_id="sess_123",
    message_id="msg_1",
    query="best laptops for programming"
)

if result["found"]:
    print("Recommendations:", result["recommendations"])

API Reference

AdMeshClient

Main client for consuming recommendations from the AdMesh Protocol service.

Constructor

client = AdMeshClient(
    api_key: str,           # Required: Your AdMesh API key
    api_base_url: str = None  # Optional: API base URL
)

All other settings (API endpoint, debug mode, timeouts) are configured internally for optimal performance.

Methods

get_recommendations_for_weave()

Async method to get recommendations for Weave format from admesh-protocol. This method makes a direct HTTP POST to /agent/recommend which either returns cached recommendations from the database or generates new ones.

result = await client.get_recommendations_for_weave(
    session_id: str,        # Session ID
    message_id: str,        # Message ID for this conversation message
    query: str = None,      # User query (optional, recommended for better recommendations)
    timeout_ms: int = None  # Max wait time (default: 12000ms)
)

# Returns:
{
    "found": bool,                           # Whether recommendations were found
    "recommendations": List[dict],           # Array of recommendations
    "query": str,                            # Original query
    "request_id": str,                       # Request ID
    "error": str                             # Error message if not found
}

Example:

result = await client.get_recommendations_for_weave(
    session_id="sess_123",
    message_id="msg_1",
    query="best project management tools",
    timeout_ms=30000
)

if result["found"]:
    print("Recommendations found:", result["recommendations"])
    print("Original query:", result["query"])

    # Use recommendations in your application
    for rec in result["recommendations"]:
        print(f"- {rec['product_title']}")
        print(f"  {rec['weave_summary']}")
        print(f"  Click: {rec['click_url']}")
        print(f"  Exposure: {rec['exposure_url']}")
else:
    print("No recommendations available:", result.get("error"))

get_recommendations_for_weave_sync()

Synchronous version of get_recommendations_for_weave(). Same parameters and return type.

result = client.get_recommendations_for_weave_sync(
    session_id="sess_123",
    message_id="msg_1",
    query="best laptops for programming"
)

Configuration

Environment Variables

# Required
ADMESH_API_KEY=your_api_key_here

# Optional - Override API base URL
ADMESH_API_BASE_URL=https://api.useadmesh.com

Initialization

The AdMeshClient requires only your API key:

client = AdMeshClient(api_key="your-api-key")

How It Works

• API Endpoint: Automatically configured to https://api.useadmesh.com
• Timeouts & Retries: Configured internally with sensible defaults

Integration Guide

Step 1: Initialize the SDK

from admesh_weave import AdMeshClient

client = AdMeshClient(api_key="your-api-key")

Step 2: Get Recommendations for Weave

async def handle_user_query(session_id, message_id, user_query):
    # Get recommendations for Weave (direct HTTP call to /agent/recommend)
    result = await client.get_recommendations_for_weave(
        session_id=session_id,
        message_id=message_id,
        query=user_query,
        timeout_ms=30000  # 30 second timeout for generation
    )

    if result["found"]:
        # Recommendations available (either from cache or freshly generated)
        print(f"Found {len(result['recommendations'])} recommendations")
        return result["recommendations"]

    # No recommendations available
    print("No recommendations:", result.get("error"))
    return None

Step 3: Use Recommendations

recommendations = await handle_user_query(session_id, message_id, query)

if recommendations:
    # Use recommendations in your application
    for rec in recommendations:
        print(f"Product: {rec['product_title']}")
        print(f"Summary: {rec['weave_summary']}")
        print(f"Click URL: {rec['click_url']}")
        print(f"Exposure URL: {rec['exposure_url']}")
        print(f"Trust Score: {rec['trust_score']}")
else:
    # Fallback behavior
    print("No recommendations available")

Performance

• Cache Hit Latency: < 100ms when recommendations are cached in database
• Cache Miss Latency: 1-3 seconds for fresh recommendation generation
• Timeout: Default 12 seconds, configurable per request
• Database TTL: Recommendations cached for 60 seconds
• Fallback: Graceful degradation if recommendations unavailable
• Lightweight: SDK is a thin HTTP client with minimal overhead

Error Handling

result = await client.get_recommendations_for_weave(
    session_id=session_id,
    message_id=message_id,
    timeout_ms=10000
)

if not result["found"]:
    print("Failed to retrieve recommendations:", result.get("error"))

    # Fallback behavior
    return {
        "success": False,
        "recommendations": [],
        "error": result.get("error")
    }

# Use recommendations
return {
    "success": True,
    "recommendations": result["recommendations"],
    "query": result["query"]
}

Troubleshooting

No recommendations found

Possible causes:

1. admesh-protocol is not running or not accessible
2. Query doesn't match any products in the database
3. Session ID or Message ID mismatch
4. Timeout too short for fresh generation
5. Invalid API key

Solution:

# Try with longer timeout
result = await client.get_recommendations_for_weave(
    session_id=session_id,
    message_id=message_id,
    query="specific product query",  # Provide a clear query
    timeout_ms=45000  # Increase timeout for generation
)

print("Found:", result["found"])
print("Error:", result.get("error"))

Connection errors

Possible causes:

1. Network connectivity issues
2. Invalid API key
3. AdMesh API service is down

Solution:

import os

# Check API key format
api_key = os.environ.get("ADMESH_API_KEY")
print(f"API key starts with: {api_key[:10] if api_key else 'None'}")

# Test the connection
try:
    result = await client.get_recommendations_for_weave(
        session_id="test",
        message_id="test",
        query="test query",
        timeout_ms=5000
    )
    print("Connection successful:", result["found"])
except Exception as error:
    print("Connection failed:", str(error))

Slow response times

Possible causes:

1. First call (cache miss) - recommendations are being generated
2. Complex query requiring more processing
3. Database connection issues

Solution:

# First call will be slower (1-3 seconds) - this is normal
result1 = await client.get_recommendations_for_weave(
    session_id=session_id,
    message_id=message_id,
    query=query,
    timeout_ms=30000
)
print("First call (generation):", result1["found"])

# Second call with same session_id + message_id should be fast (< 100ms)
result2 = await client.get_recommendations_for_weave(
    session_id=session_id,
    message_id=message_id,
    query=query,
    timeout_ms=5000  # Can use shorter timeout for cached results
)
print("Second call (cache hit):", result2["found"])

Types

AdMeshRecommendation

Individual recommendation object returned by the API.

{
    "ad_id": str,                          # Unique ad identifier
    "product_id": str,                     # Product ID
    "recommendation_id": str,              # Recommendation identifier
    "product_title": str,                  # Product name
    "citation_summary": str,               # Citation text
    "weave_summary": str,                  # Weave format summary
    "exposure_url": str,                   # Exposure tracking URL
    "click_url": str,                      # Click tracking URL
    "product_logo": {"url": str},          # Product logo object
    "categories": List[str],               # Product categories
    "contextual_relevance_score": float,   # Contextual relevance score (0-100)
    "trust_score": float,                  # Trust score
    "model_used": str                      # Model used for generation
}

AdMeshWaitResult

Result from get_recommendations_for_weave() method.

{
    "found": bool,                         # Whether recommendations were found
    "recommendations": List[dict],         # Array of recommendations
    "query": str,                          # Original query
    "request_id": str,                     # Request ID
    "error": str                           # Error message if not found
}

AdMeshClientConfig

Configuration for initializing the AdMeshClient.

{
    "api_key": str,        # Required: Your AdMesh API key
    "api_base_url": str    # Optional: API base URL
}

Architecture Details

How It Works

1. admesh-weave-python SDK makes HTTP POST to /agent/recommend endpoint
2. admesh-protocol checks database cache for existing recommendations
3. Cache Hit: Returns cached recommendations immediately (< 100ms)
4. Cache Miss: Generates new recommendations, saves to database, returns (1-3s)
5. SDK returns recommendations in a format ready for LLM integration

Data Flow

Your Application
    ↓
admesh-weave-python SDK
    ↓ (HTTP POST)
admesh-protocol /agent/recommend
    ↓
Check Database Cache (session_id + message_id)
    ↓
Cache Hit? → Return Cached (< 100ms)
    ↓ (cache miss)
Generate Recommendations
    ↓
Save to Database (60s TTL)
    ↓
Return Recommendations
    ↓
SDK Formats for LLM
    ↓
Integrate into Prompt
    ↓
LLM Response with Weave Ads

Database Caching

• Collection: recommendations in Firestore
• Query: By session_id and message_id fields
• Freshness Window: 60 seconds (validated using created_at timestamp)
• Cache Hit: < 100ms response time
• Cache Miss: 1-3 seconds for fresh generation
• No TTL Required: Application-level freshness validation

Key Design Principles

• Lightweight: SDK is a thin HTTP client with minimal dependencies
• Fast: Database cache hits return in < 100ms
• Simple: Direct request/response pattern, no complex subscriptions
• Reliable: No message delivery issues, no race conditions
• Graceful Fallback: Works seamlessly even if recommendations aren't available
• Configurable: Timeout is customizable per request
• Type-safe: Full type hints for better IDE support

Requirements

• Python 3.8+
• httpx >= 0.23.0
• typing-extensions >= 4.5.0

Development

Running Tests

# Install development dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run tests with coverage
pytest --cov=admesh_weave

Code Quality

# Format code
ruff format

# Lint code
ruff check .

# Fix linting issues
ruff check --fix .

# Type check
mypy .

License

MIT