RubiPro Agent API v2

The RubiPro Agent API v2 is a modern REST API designed to enable AI agents to integrate seamlessly with the RubiPro CRM system. AI agents are first-class citizens alongside human users—they have their own identity, authentication, permissions, and full audit trails of all actions.

What is an AI Agent?

In RubiPro, an AI agent is an autonomous application that can:

• Authenticate independently using OAuth2 Client Credentials
• Read and write customer data with granular permission controls
• Log interactions and annotations on customer records
• Perform bulk operations across multiple customers
• Generate reports and analysis
• Escalate to human agents when needed
• Leave audit trails of every action taken

Agents can be built with popular AI frameworks like LangChain, Claude, GPT-4, or as custom microservices. They can be internal (RubiPro-maintained) or external (customer-built).

Quick Start

Get your first agent running in 5 minutes.

Step 1: Register Your Agent

1. Log in to your tenant admin panel
2. Navigate to Security → Agents (/tenant-admin/agents.php)
3. Click Register New Agent
4. Enter agent name (e.g., "My Customer Assistant")
5. Select permissions (start with customers:read)
6. Click Create Agent

Step 2: Capture Credentials

You'll receive:

• client_id — Your agent's unique identifier
• client_secret — Keep this secret! (shown only once)

Store these securely in your environment variables or secret manager.

Step 3: Get an Access Token

import requests
import json

# Exchange credentials for access token
response = requests.post(
    'https://rubiprofessional.com/api/v2/agent/auth/token.php',
    json={
        'client_id': 'YOUR_CLIENT_ID',
        'client_secret': 'YOUR_CLIENT_SECRET',
        'grant_type': 'client_credentials'
    }
)

token_data = response.json()
access_token = token_data['access_token']
print(f"Token expires in: {token_data['expires_in']} seconds")

const response = await fetch(
  'https://rubiprofessional.com/api/v2/agent/auth/token.php',
  {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      client_id: process.env.AGENT_CLIENT_ID,
      client_secret: process.env.AGENT_CLIENT_SECRET,
      grant_type: 'client_credentials'
    })
  }
);

const { access_token, expires_in } = await response.json();
console.log(`Token valid for ${expires_in} seconds`);

Step 4: Make Your First API Call

import requests

headers = {
    'Authorization': f'Bearer {access_token}',
    'Content-Type': 'application/json'
}

# Get list of customers
response = requests.get(
    'https://rubiprofessional.com/api/v2/customers.php',
    headers=headers,
    params={'limit': 10}
)

customers = response.json()
for customer in customers['data']:
    print(f"{customer['first_name']} {customer['last_name']}")
    print(f"  Email: {customer['email']}")

// List customers
const customersResponse = await fetch(
  'https://rubiprofessional.com/api/v2/customers.php?limit=10',
  {
    headers: {
      'Authorization': `Bearer ${access_token}`,
      'Content-Type': 'application/json'
    }
  }
);

const { data: customers } = await customersResponse.json();
customers.forEach(customer => {
  console.log(`${customer.first_name} ${customer.last_name}`);
  console.log(`  Email: ${customer.email}`);
});

Authentication

RubiPro uses OAuth2 Client Credentials flow for agent authentication. This is the standard way for server-to-server communication without user interaction.

Token Endpoint

POST https://rubiprofessional.com/api/v2/agent/auth/token.php

Request Body

Response

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "Bearer",
  "expires_in": 3600,
  "refresh_token": "refresh_token_here"
}

Using the Access Token

Include the token in all API requests as a Bearer token in the Authorization header:

curl -X GET https://rubiprofessional.com/api/v2/customers.php \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json"

Token Refresh

Access tokens expire in 1 hour. Use the refresh token to get a new access token without re-authenticating:

POST https://rubiprofessional.com/api/v2/agent/auth/refresh.php

{
  "refresh_token": "your_refresh_token"
}

Token Revocation

Revoke a token immediately (useful for security events):

POST https://rubiprofessional.com/api/v2/agent/auth/revoke.php

{
  "token": "access_token_to_revoke"
}

Token Introspection

Check token validity and remaining scope:

POST https://rubiprofessional.com/api/v2/agent/auth/introspect.php

{
  "token": "access_token_to_check"
}

API Endpoints

Complete reference of all available Agent API v2 endpoints.

Authentication

Agent Management

Customers

Interactions

Advanced Features

Permissions & Scopes

RubiPro uses scoped permissions to control what your agent can do. Permissions are granular and field-aware, with optional conditions for advanced access control.

Available Scopes

Field-Level Access Control

Your agent can be restricted to specific customer fields. For example, agents handling public inquiries might only see name, email, and phone—not custom VIP or internal fields.

Blocked Actions (Always)

These actions are permanently blocked for all agents, regardless of scope:

• Manage users or roles
• Manage other agents
• Access billing information
• Modify system settings
• Delete customer records (unless explicitly granted)
• Export sensitive data (unless explicitly granted)

Conditional Access

Advanced use case: Restrict an agent to only customers matching certain criteria (e.g., status='hot_lead'). Use conditions when setting up agent permissions.

Handoffs & Human-in-the-Loop

Agents can escalate conversations to humans when needed. The handoff system includes an approval queue for sensitive operations.

Triggering a Handoff

import requests

headers = {'Authorization': f'Bearer {access_token}'}

# Request human escalation
response = requests.post(
    'https://rubiprofessional.com/api/v2/handoff.php',
    headers=headers,
    json={
        'customer_id': 456,
        'reason': 'Customer requesting manager approval',
        'priority': 'high',
        'context': 'Customer wants to negotiate contract terms'
    }
)

handoff = response.json()
print(f"Handoff ticket: {handoff['ticket_id']}")
print(f"Status: {handoff['status']}")

Checking Approval Queue

Monitor pending actions awaiting human review:

response = requests.get(
    'https://rubiprofessional.com/api/v2/pending-actions.php',
    headers=headers
)

pending = response.json()
for action in pending['data']:
    print(f"Action: {action['action_type']}")
    print(f"Customer: {action['customer_name']}")
    print(f"Status: {action['status']}")
    print(f"Requested by agent: {action['agent_name']}")
    print("---")

Handoff States

• pending — Awaiting human review
• approved — Human approved the action
• rejected — Human rejected the action
• escalated — Escalated to higher authority

Rate Limiting

RubiPro enforces sliding-window rate limiting to protect system stability. Limits vary by endpoint and scope.

Default Limits

Rate Limit Headers

Every response includes rate limit information:

X-RateLimit-Limit: 1000
X-RateLimit-Remaining: 847
X-RateLimit-Reset: 1710329400
X-RateLimit-RetryAfter: 3600

Handling Rate Limit Errors

When you exceed limits, you'll receive a 429 Too Many Requests response. Implement exponential backoff:

import time
import requests

def make_request_with_backoff(url, headers, max_retries=3):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers)

        if response.status_code == 429:
            retry_after = int(response.headers.get('X-RateLimit-RetryAfter', 60))
            wait_time = retry_after * (2 ** attempt)  # Exponential backoff
            print(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
            continue

        return response

    raise Exception("Max retries exceeded")

response = make_request_with_backoff(url, headers)

Error Handling

RubiPro returns standard HTTP status codes and structured error responses.

Error Response Format

{
  "error": "invalid_request",
  "error_description": "Missing required parameter: client_id",
  "error_code": 400,
  "request_id": "req_1234567890"
}

Common Error Codes

Best Practices

• Always check HTTP status code first
• Log the request_id for debugging support tickets
• Implement retry logic for 5xx errors
• Never retry 4xx errors (except 429)
• Parse error_description for user-facing messages

SDKs & Examples

Code examples for common tasks using Python and JavaScript/Node.js.

List Customers with Pagination

import requests

headers = {'Authorization': f'Bearer {access_token}'}

# Fetch first page
response = requests.get(
    'https://rubiprofessional.com/api/v2/customers.php',
    headers=headers,
    params={
        'limit': 50,
        'offset': 0
    }
)

data = response.json()
total = data['pagination']['total']
print(f"Total customers: {total}")

for customer in data['data']:
    print(f"{customer['first_name']} {customer['last_name']}")

# Check if more pages exist
if data['pagination']['has_more']:
    print("More customers available—fetch next page")

Create a Customer

const response = await fetch(
  'https://rubiprofessional.com/api/v2/customers.php',
  {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${access_token}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      first_name: 'John',
      last_name: 'Doe',
      email: 'john@example.com',
      phone: '+1-555-0123',
      status: 'active'
    })
  }
);

const customer = await response.json();
console.log(`Created customer ${customer.id}`);

Log an Interaction

response = requests.post(
    'https://rubiprofessional.com/api/v2/interactions.php',
    headers=headers,
    json={
        'customer_id': 123,
        'type': 'email',
        'direction': 'outbound',
        'subject': 'Follow-up on your inquiry',
        'notes': 'Customer responded positively to proposal',
        'metadata': {
            'sentiment': 'positive',
            'duration': 240,
            'resolution': True
        }
    }
)

interaction = response.json()
print(f"Logged interaction {interaction['id']}")

Bulk Update Customers

const response = await fetch(
  'https://rubiprofessional.com/api/v2/bulk.php',
  {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${access_token}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      action: 'update',
      customer_ids: [123, 456, 789],
      updates: {
        status: 'qualified',
        assigned_to: 'sales_team'
      }
    })
  }
);

const { task_id, records_affected } = await response.json();
console.log(`Updated ${records_affected} customers`);

Search Customers

response = requests.get(
    'https://rubiprofessional.com/api/v2/search.php',
    headers=headers,
    params={
        'q': 'acme corp',
        'fields': ['first_name', 'last_name', 'company'],
        'limit': 10
    }
)

results = response.json()
for match in results['matches']:
    print(f"Score: {match['score']}")
    print(f"Customer: {match['data']['first_name']} {match['data']['last_name']}")

Security Best Practices

1. Secret Management

• Never commit secrets to version control — Use environment variables or secret managers
• Rotate credentials regularly — Regenerate client secrets every 90 days
• Use a secrets manager — Consider AWS Secrets Manager, HashiCorp Vault, or similar
• Audit access logs — Monitor who accessed secrets

2. Token Handling

• Keep tokens in backend only — Never expose in frontend or URLs
• Use HTTPS everywhere — Prevent token interception
• Implement token refresh — Don't rely on long-lived tokens
• Validate token expiry — Check expires_in before making requests

3. Request Validation

• Verify HTTPS certificates — Disable certificate verification only in development
• Validate API responses — Never trust unvalidated data
• Use request signing — Include request IDs for idempotency
• Implement input validation — Sanitize all inputs before sending

4. Error Handling

• Never expose stack traces — Log internally, return generic messages
• Log request IDs — Include in error handling for debugging
• Monitor error rates — Alert on unusual patterns
• Implement rate limit handling — Respect backoff signals

5. Monitoring & Alerts

• Track API usage — Monitor for unusual patterns
• Alert on failures — Set up notifications for 401, 403, 429 errors
• Review audit logs — Check /tenant-admin/agents.php for agent activity
• Enable anomaly detection — Let RubiPro flag suspicious behavior

6. Testing

# Always test error cases
def test_unauthorized():
    response = requests.get(
        'https://rubiprofessional.com/api/v2/customers.php',
        headers={'Authorization': 'Bearer invalid_token'}
    )
    assert response.status_code == 401

def test_rate_limiting():
    for i in range(1005):  # Exceed limit
        response = requests.get(url, headers=headers)

    assert response.status_code == 429
    assert 'X-RateLimit-RetryAfter' in response.headers

# Run comprehensive security tests
if __name__ == '__main__':
    test_unauthorized()
    test_rate_limiting()
    print("Security tests passed!")