ContractAI Integration Guide

Complete guide to integrating ContractAI with your systems and services

Overview

This guide provides comprehensive instructions for integrating ContractAI with your existing infrastructure, applications, and services. It covers API integration, third-party service connections, and best practices for building robust integrations.

Integration Architecture

graph TD
    A[Integration Layer] --> B[API Gateway]
    A --> C[Event Bus]
    A --> D[Message Queue]

    B --> B1[REST API]
    B --> B2[GraphQL]
    B --> B3[WebSocket]

    C --> C1[Events]
    C --> C2[Notifications]
    C --> C3[Webhooks]

    D --> D1[Tasks]
    D --> D2[Commands]
    D --> D3[Updates]

Integration Patterns

System Integration

graph TD
    A[Your System] --> B[Integration Methods]
    B --> C[Direct API]
    B --> D[Event-Based]
    B --> E[Message-Based]

    C --> C1[REST]
    C --> C2[GraphQL]
    C --> C3[gRPC]

    D --> D1[Webhooks]
    D --> D2[Pub/Sub]
    D --> D3[Streaming]

    E --> E1[Queue]
    E --> E2[Topics]
    E --> E3[Streams]

Data Flow

sequenceDiagram
    participant S as System
    participant I as Integration
    participant C as ContractAI
    participant D as Data Store

    S->>I: Send Data
    I->>C: Transform & Forward
    C->>D: Store
    D->>C: Confirm
    C->>I: Process
    I->>S: Update Status

1. Integration Types

Available Integration Methods

# Integration Options
integration_methods:
  - name: "REST API"
    use_case: "Direct system integration"
    features:
      - CRUD operations
      - Real-time updates
      - Webhook support

  - name: "SDK"
    use_case: "Application development"
    features:
      - Type-safe operations
      - Simplified authentication
      - Built-in retry logic

  - name: "Webhooks"
    use_case: "Event-driven integration"
    features:
      - Real-time notifications
      - Custom event handling
      - Retry mechanisms

  - name: "Message Queue"
    use_case: "Asynchronous processing"
    features:
      - Reliable delivery
      - High throughput
      - Message persistence

Integration Architecture

graph TD
    A[Your System] -->|REST API| B[ContractAI API Gateway]
    A -->|SDK| B
    B -->|Webhooks| A
    B -->|Message Queue| C[Event Processor]
    C -->|Processed Events| A
    B -->|Database| D[(Knowledge Base)]
    B -->|Cache| E[(Redis Cache)]

2. Integration Requirements

System Requirements

• Python 3.12 or later
• Network access to ContractAI services
• Valid API credentials
• Minimum 1GB RAM for SDK
• Stable internet connection

Security Requirements

• TLS 1.3 for all connections
• API key or OAuth 2.0 credentials
• IP allowlisting (optional)
• MFA for admin operations

API Integration

Authentication Flow

sequenceDiagram
    participant C as Client
    participant A as Auth
    participant G as Gateway
    participant S as Service

    C->>A: Request Token
    A->>C: Return Token
    C->>G: API Request + Token
    G->>A: Validate Token
    A->>G: Token Valid
    G->>S: Process Request
    S->>C: Return Response

API Request Flow

graph TD
    A[API Request] --> B[Authentication]
    A --> C[Validation]
    A --> D[Processing]

    B --> B1[Token Check]
    B --> B2[Permission]
    B --> B3[Rate Limit]

    C --> C1[Input]
    C --> C2[Schema]
    C --> C3[Business Rules]

    D --> D1[Execute]
    D --> D2[Transform]
    D --> D3[Response]

1. Authentication

API Key Authentication

# Using API Key
import requests
from contractai import ContractAIClient

# Initialize client
client = ContractAIClient(
    api_key="your_api_key",
    environment="production"
)

# Make authenticated request
response = client.agents.list()

OAuth 2.0 Authentication

# Using OAuth 2.0
from contractai import ContractAIClient
from oauthlib.oauth2 import BackendApplicationClient

# Initialize OAuth client
oauth_client = BackendApplicationClient(
    client_id="your_client_id"
)

# Initialize ContractAI client
client = ContractAIClient(
    oauth_client=oauth_client,
    environment="production"
)

# Authenticate and make request
client.authenticate()
response = client.agents.list()

2. API Endpoints

Agent Management

# Agent Operations
from contractai import ContractAIClient

client = ContractAIClient(api_key="your_api_key")

# Create agent
agent = client.agents.create(
    name="production-monitor",
    capabilities=["monitoring", "cost-optimization"]
)

# Get agent status
status = client.agents.get_status(agent_id=agent.id)

# Update agent
client.agents.update(
    agent_id=agent.id,
    capabilities=["monitoring", "cost-optimization", "security-scanning"]
)

# Delete agent
client.agents.delete(agent_id=agent.id)

Knowledge Base Integration

# Knowledge Base Operations
from contractai import ContractAIClient

client = ContractAIClient(api_key="your_api_key")

# Add document
document = client.knowledge.add_document(
    content="System architecture documentation",
    category="architecture",
    tags=["system", "design"]
)

# Query knowledge base
results = client.knowledge.query(
    query="How to handle high CPU usage?",
    context={"environment": "production"}
)

# Update document
client.knowledge.update_document(
    document_id=document.id,
    content="Updated system architecture",
    version="2.0"
)

3. Webhook Integration

Webhook Setup

# Webhook Configuration
from contractai import ContractAIClient

client = ContractAIClient(api_key="your_api_key")

# Register webhook
webhook = client.webhooks.create(
    url="https://your-system.com/webhooks/contractai",
    events=["agent.status_change", "incident.created"],
    secret="your_webhook_secret"
)

# Verify webhook
client.webhooks.verify(webhook_id=webhook.id)

# List webhooks
webhooks = client.webhooks.list()

Webhook Handler

# FastAPI Webhook Handler
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()

@app.post("/webhooks/contractai")
async def handle_webhook(request: Request):
    # Verify webhook signature
    signature = request.headers.get("X-ContractAI-Signature")
    payload = await request.body()

    expected_signature = hmac.new(
        "your_webhook_secret".encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected_signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Process webhook
    data = await request.json()
    event_type = data["event_type"]

    if event_type == "agent.status_change":
        handle_agent_status_change(data)
    elif event_type == "incident.created":
        handle_incident_created(data)

    return {"status": "success"}

Third-Party Integration

1. Cloud Provider Integration

AWS Integration

# AWS Integration
from contractai import ContractAIClient
import boto3

# Initialize clients
contract_client = ContractAIClient(api_key="your_api_key")
aws_client = boto3.client("ec2")

# Create AWS-aware agent
agent = contract_client.agents.create(
    name="aws-monitor",
    capabilities=["aws-monitoring"],
    config={
        "aws_region": "us-west-2",
        "aws_services": ["ec2", "rds", "lambda"]
    }
)

# Monitor AWS resources
def monitor_aws_resources():
    instances = aws_client.describe_instances()
    for instance in instances["Reservations"]:
        contract_client.monitoring.record_metric(
            agent_id=agent.id,
            metric="aws.ec2.instance_status",
            value=instance["State"]["Name"]
        )

Azure Integration

# Azure Integration
from contractai import ContractAIClient
from azure.mgmt.compute import ComputeManagementClient
from azure.identity import DefaultAzureCredential

# Initialize clients
contract_client = ContractAIClient(api_key="your_api_key")
azure_credential = DefaultAzureCredential()
azure_client = ComputeManagementClient(
    credential=azure_credential,
    subscription_id="your_subscription_id"
)

# Create Azure-aware agent
agent = contract_client.agents.create(
    name="azure-monitor",
    capabilities=["azure-monitoring"],
    config={
        "azure_subscription": "your_subscription_id",
        "azure_services": ["compute", "storage", "network"]
    }
)

2. Monitoring Integration

Prometheus Integration

# Prometheus Integration
from contractai import ContractAIClient
from prometheus_client import start_http_server, Gauge

# Initialize client
client = ContractAIClient(api_key="your_api_key")

# Create Prometheus metrics
agent_status = Gauge(
    "contractai_agent_status",
    "Status of ContractAI agents",
    ["agent_id", "agent_name"]
)

# Start Prometheus server
start_http_server(8000)

# Update metrics
def update_metrics():
    agents = client.agents.list()
    for agent in agents:
        status = client.agents.get_status(agent.id)
        agent_status.labels(
            agent_id=agent.id,
            agent_name=agent.name
        ).set(1 if status.healthy else 0)

Grafana Integration

# Grafana Dashboard Configuration
apiVersion: 1

dashboards:
  - name: "ContractAI Overview"
    uid: "contractai-overview"
    panels:
      - title: "Agent Status"
        type: "stat"
        targets:
          - expr: "contractai_agent_status"
            legendFormat: "{{agent_name}}"

      - title: "Incident Rate"
        type: "graph"
        targets:
          - expr: "rate(contractai_incidents_total[5m])"
            legendFormat: "Incidents per second"

3. CI/CD Integration

GitHub Actions Integration

# GitHub Actions Workflow
name: ContractAI Integration

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Deploy to ContractAI
        uses: contractai/deploy-action@v1
        with:
          api_key: ${{ secrets.CONTRACT_AI_API_KEY }}
          environment: production
          agent_name: "github-deploy-agent"

Jenkins Integration

// Jenkins Pipeline
pipeline {
    agent any

    environment {
        CONTRACT_AI_API_KEY = credentials('contract-ai-api-key')
    }

    stages {
        stage('Deploy to ContractAI') {
            steps {
                script {
                    def client = new ContractAIClient(
                        apiKey: env.CONTRACT_AI_API_KEY
                    )

                    client.agents.deploy(
                        name: "jenkins-deploy-agent",
                        environment: "production"
                    )
                }
            }
        }
    }
}

Best Practices

1. Integration Design

Design Principles

• Use asynchronous operations where possible
• Implement proper error handling
• Follow security best practices
• Monitor integration health
• Maintain documentation

Error Handling

# Error Handling Example
from contractai import ContractAIClient, ContractAIError
from tenacity import retry, stop_after_attempt, wait_exponential

client = ContractAIClient(api_key="your_api_key")

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def safe_agent_operation(agent_id: str):
    try:
        return client.agents.get_status(agent_id)
    except ContractAIError as e:
        if e.status_code == 429:  # Rate limit
            raise  # Retry
        elif e.status_code == 404:  # Not found
            raise  # Don't retry
        else:
            # Log and handle other errors
            logger.error(f"Operation failed: {e}")
            raise

2. Performance Optimization

Caching Strategy

# Caching Example
from contractai import ContractAIClient
from functools import lru_cache
import redis

# Initialize clients
contract_client = ContractAIClient(api_key="your_api_key")
redis_client = redis.Redis(host="localhost", port=6379)

@lru_cache(maxsize=100)
def get_agent_status(agent_id: str):
    # Try cache first
    cached_status = redis_client.get(f"agent_status:{agent_id}")
    if cached_status:
        return cached_status

    # Get from API
    status = contract_client.agents.get_status(agent_id)

    # Cache result
    redis_client.setex(
        f"agent_status:{agent_id}",
        300,  # 5 minutes
        status.json()
    )

    return status

Rate Limiting

# Rate Limiting Example
from contractai import ContractAIClient
from ratelimit import limits, sleep_and_retry

client = ContractAIClient(api_key="your_api_key")

# Limit to 100 calls per minute
@sleep_and_retry
@limits(calls=100, period=60)
def rate_limited_operation():
    return client.agents.list()

Troubleshooting

1. Common Issues

Connection Issues

• Verify network connectivity
• Check API credentials
• Validate IP allowlist
• Review firewall rules

Authentication Issues

• Verify API key validity
• Check OAuth token expiration
• Validate webhook signatures
• Review access permissions

2. Debugging Tools

Logging

# Logging Configuration
import logging
from contractai import ContractAIClient

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

# Enable debug logging
client = ContractAIClient(
    api_key="your_api_key",
    debug=True
)

Monitoring

# Integration Monitoring
from contractai import ContractAIClient
from prometheus_client import Counter, Histogram

# Define metrics
api_calls = Counter(
    "contractai_api_calls_total",
    "Total API calls",
    ["endpoint", "status"]
)

api_latency = Histogram(
    "contractai_api_latency_seconds",
    "API call latency",
    ["endpoint"]
)

# Monitor API calls
def monitored_api_call(endpoint: str):
    with api_latency.labels(endpoint=endpoint).time():
        try:
            response = client.api.call(endpoint)
            api_calls.labels(
                endpoint=endpoint,
                status="success"
            ).inc()
            return response
        except Exception as e:
            api_calls.labels(
                endpoint=endpoint,
                status="error"
            ).inc()
            raise

Additional Resources

1. Integration Tools

• API Documentation
• SDK Documentation
• Integration Examples
• Webhook Guide

2. Support Resources

• Integration Support
• Developer Community
• Knowledge Base
• Status Page

---

Need help with integration? Contact our integration team at [email protected] or visit our Integration Portal