Integration with External APIs

Cleo provides flexibility for interacting with external APIs to extend its capabilities. Whether you need to pull data from external sources, trigger external actions, or connect to third-party systems, Cleo offers simple ways to integrate with RESTful APIs, webhooks, and other external services.

This guide covers setting up Cleo to communicate with APIs, handling data exchange, and integrating external services into your agent workflows.

---

Prerequisites

Before integrating Cleo with an external API, ensure you have the following:

• Access to the external API, including the API key or credentials.

• Knowledge of the API’s endpoints and methods (GET, POST, PUT, DELETE).

• Familiarity with HTTP requests and response handling in Python.

• Cleo’s core environment set up and running.

---

Setting Up External API Communication

1. Install Requests Library

Cleo uses Python's requests library to make API calls. If it's not already installed in your environment, install it via pip:

bashCopyEditpip install requests

2. Example: Simple GET Request

Here’s a simple example of how to make a GET request to an external API to fetch data:

pythonCopyEditimport requests

def fetch_external_data():
    url = "https://api.example.com/data"
    headers = {
        "Authorization": "Bearer YOUR_API_KEY"
    }
    
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        data = response.json()
        return data
    else:
        print(f"Failed to fetch data: {response.status_code}")
        return None

In this example, the function fetch_external_data sends a GET request to https://api.example.com/data, retrieves the JSON response, and handles any errors if the request fails.

3. Sending Data with POST Requests

You can send data to an API using a POST request. For example, if you want to submit agent activity data to an external service:

pythonCopyEditdef send_agent_data(agent_id, activity_data):
    url = "https://api.example.com/submit_activity"
    headers = {
        "Authorization": "Bearer YOUR_API_KEY",
        "Content-Type": "application/json"
    }
    payload = {
        "agent_id": agent_id,
        "activity_data": activity_data
    }
    
    response = requests.post(url, json=payload, headers=headers)
    
    if response.status_code == 200:
        print("Data sent successfully")
    else:
        print(f"Failed to send data: {response.status_code}")

This function sends a POST request with a JSON payload to an API endpoint, passing agent activity data.

---

Handling API Responses

1. Error Handling

Proper error handling ensures that Cleo can handle failure scenarios gracefully. You can check the status code and handle different response codes appropriately:

pythonCopyEditdef handle_api_response(response):
    if response.status_code == 200:
        return response.json()  # Successful request
    elif response.status_code == 400:
        print("Bad request")
    elif response.status_code == 401:
        print("Unauthorized access")
    elif response.status_code == 500:
        print("Internal server error")
    else:
        print(f"Unexpected error: {response.status_code}")

By checking the response status code, you can adapt the agent’s behavior depending on the API’s response.

2. Parsing JSON Responses

Most APIs return data in JSON format. Cleo can easily handle this using Python’s built-in json module or directly via requests:

pythonCopyEditresponse_data = response.json()  # Parses JSON into a Python dictionary

Once parsed, you can access the data as a standard Python dictionary:

pythonCopyEditdata_value = response_data['key']

---

Authentication Methods

External APIs often require authentication, and Cleo supports various methods to authenticate API calls.

1. API Key Authentication

The simplest authentication method is using an API key, which is passed in the request headers:

pythonCopyEditheaders = {
    "Authorization": "Bearer YOUR_API_KEY"
}

2. OAuth 2.0 Authentication

For more secure authentication, APIs may require OAuth 2.0. Cleo can handle OAuth 2.0 tokens by following the OAuth flow to obtain the access token, typically by using an authentication library such as requests_oauthlib:

bashCopyEditpip install requests_oauthlib

Example OAuth 2.0 Integration:

pythonCopyEditfrom requests_oauthlib import OAuth2Session

# OAuth2 details
client_id = 'YOUR_CLIENT_ID'
client_secret = 'YOUR_CLIENT_SECRET'
authorization_base_url = 'https://api.example.com/oauth/authorize'
token_url = 'https://api.example.com/oauth/token'

# Create an OAuth2 session
oauth = OAuth2Session(client_id)
authorization_url, state = oauth.authorization_url(authorization_base_url)

# The user will need to visit the authorization URL
print(f"Please go to {authorization_url} and authorize access.")

# After authorization, obtain the token
oauth.fetch_token(token_url, client_secret=client_secret, authorization_response=input('Enter the full callback URL: '))

# Make API requests using OAuth2 session
response = oauth.get('https://api.example.com/secure-data')
print(response.json())

---

Scheduling API Calls

To make periodic API calls, you can integrate the requests library with APScheduler or any task scheduler to make recurring API calls automatically. Here's an example using APScheduler to fetch data from an API every hour:

pythonCopyEditfrom apscheduler.schedulers.blocking import BlockingScheduler

def scheduled_api_call():
    url = "https://api.example.com/endpoint"
    response = requests.get(url)
    if response.status_code == 200:
        print("Data received:", response.json())
    else:
        print(f"Error: {response.status_code}")

scheduler = BlockingScheduler()
scheduler.add_job(scheduled_api_call, 'interval', hours=1)
scheduler.start()

---

Webhook Integration

If your external service supports webhooks, you can set up Cleo to receive real-time updates or data via HTTP POST requests. To handle webhooks in Cleo:

1. Set up a web server (e.g., using Flask or FastAPI) to listen for incoming POST requests.

2. Define an endpoint where the external service will send the data.

Example with Flask:

pythonCopyEditfrom flask import Flask, request

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.json
    print("Received webhook data:", data)
    # Process the data
    return "Webhook received", 200

if __name__ == '__main__':
    app.run(debug=True, port=5000)

Make sure the external API sends POST requests to the /webhook endpoint when an event occurs.

---

Summary

Integrating Cleo with external APIs opens up a world of possibilities for your agents. Whether pulling data, sending actions, or receiving updates via webhooks, Cleo can be easily configured to interact with any RESTful API. By using libraries like requests and APScheduler, you can create seamless integrations that extend the functionality and autonomy of your Cleo agents.