API Integration

# Example: Petstore API
- petstore:
    url: http://localhost:8080/openapi.json
    type: openapi
    auth:
      type: none  # or "bearer", "basic", "api_key"

# Example: Custom API with authentication
- myapi:
    url: https://api.example.com/openapi.json
    type: openapi
    auth:
      type: bearer
      value: Bearer your-token-here

[apis.petstore]
# API-specific settings
timeout = 30
max_retries = 3
cache_enabled = true
cache_ttl = 300

[apis.myapi]
# Custom API settings
timeout = 60
max_retries = 5
rate_limit = 100  # requests per minute

# Check registered APIs
cuga apis list

# Test API connectivity
cuga apis test petstore

# Show API details
cuga apis info petstore

# Stop existing container
docker stop petstore 2>/dev/null
docker rm petstore 2>/dev/null

# Pull and run Petstore
docker pull openapitools/openapi-petstore
docker run -d \
  -e OPENAPI_BASE_PATH=/v3 \
  -e DISABLE_API_KEY=1 \
  -e DISABLE_OAUTH=1 \
  -p 8080:8080 \
  --name petstore \
  openapitools/openapi-petstore

# agent/api/config/mcp_servers.yaml
- petstore:
    url: http://localhost:8080/openapi.json
    type: openapi
    auth:
      type: none

# Start CUGA demo
cuga start demo

# Try queries like:
"show me all available pets in the petstore"
"add a new pet named Fluffy to the store"
"find pets by status available"

# my_tools.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr
from typing import Dict, List

app = FastAPI(title="My Tools API")

# Sample data
contacts_db = {
    "john@example.com": {"name": "John Doe", "role": "Developer"},
    "jane@example.com": {"name": "Jane Smith", "role": "Manager"}
}

class Contact(BaseModel):
    name: str
    role: str

@app.get("/contacts/{email}")
def get_contact(email: EmailStr):
    """
    Retrieve details of a contact identified by their email.
    """
    if email not in contacts_db:
        raise HTTPException(404, "Contact not found")
    return contacts_db[email]

@app.get("/contacts")
def list_contacts():
    """
    List all available contacts.
    """
    return list(contacts_db.values())

@app.post("/contacts")
def create_contact(contact: Contact):
    """
    Create a new contact.
    """
    # Implementation here
    return {"message": "Contact created", "contact": contact}

# Install dependencies
pip install fastapi uvicorn

# Run server
uvicorn my_tools:app --host 0.0.0.0 --port 8111 --reload

# agent/api/config/mcp_servers.yaml
- mytools:
    url: http://localhost:8111/openapi.json
    type: openapi
    auth:
      type: none

# Start CUGA
cuga start demo

# Try queries like:
"get contact information for john@example.com"
"list all available contacts"
"create a new contact for Alice Johnson"

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    if credentials.credentials != "valid-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    return credentials.credentials

@app.get("/secure/contacts")
def get_secure_contacts(token: str = Depends(verify_token)):
    return list(contacts_db.values())

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/contacts")
@limiter.limit("10/minute")
def list_contacts(request: Request):
    return list(contacts_db.values())

# agent/api/config/mcp_servers.yaml
- my_mcp_server:
    url: localhost:8000
    type: mcp
    auth:
      type: none
    tools:
      - get_weather
      - calculate_distance
      - send_notification

# mcp_server.py
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent

server = Server("my-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="Get current weather for a location",
            inputSchema={
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_weather":
        location = arguments.get("location", "Unknown")
        return [TextContent(type="text", text=f"Weather in {location}: Sunny")]
    
    return [TextContent(type="text", text="Tool not found")]

if __name__ == "__main__":
    asyncio.run(server.run())

# custom_tools.py
from typing import Dict, Any

def calculate_revenue(accounts: list, multiplier: float = 1.0) -> Dict[str, Any]:
    """
    Calculate total revenue from a list of accounts.
    
    Args:
        accounts: List of account dictionaries
        multiplier: Revenue multiplier factor
    
    Returns:
        Dictionary with revenue calculations
    """
    total_revenue = sum(account.get('revenue', 0) for account in accounts)
    adjusted_revenue = total_revenue * multiplier
    
    return {
        "total_revenue": total_revenue,
        "adjusted_revenue": adjusted_revenue,
        "account_count": len(accounts),
        "multiplier": multiplier
    }

def analyze_performance(accounts: list) -> Dict[str, Any]:
    """
    Analyze account performance metrics.
    """
    if not accounts:
        return {"error": "No accounts provided"}
    
    revenues = [acc.get('revenue', 0) for acc in accounts]
    return {
        "total_revenue": sum(revenues),
        "average_revenue": sum(revenues) / len(revenues),
        "max_revenue": max(revenues),
        "min_revenue": min(revenues),
        "top_performers": sorted(accounts, key=lambda x: x.get('revenue', 0), reverse=True)[:3]
    }

# agent/api/config/mcp_servers.yaml
- custom_tools:
    type: python
    module: custom_tools
    functions:
      - calculate_revenue
      - analyze_performance

# Start CUGA
cuga start demo

# Try queries like:
"calculate revenue for all accounts with a 1.2 multiplier"
"analyze the performance of our top accounts"

- secure_api:
    url: https://api.secure.com/openapi.json
    type: openapi
    auth:
      type: bearer
      value: Bearer your-token-here

- api_key_api:
    url: https://api.example.com/openapi.json
    type: openapi
    auth:
      type: api_key
      name: X-API-Key
      value: your-api-key-here

- basic_auth_api:
    url: https://api.example.com/openapi.json
    type: openapi
    auth:
      type: basic
      username: your-username
      password: your-password

- oauth_api:
    url: https://api.example.com/openapi.json
    type: openapi
    auth:
      type: oauth2
      client_id: your-client-id
      client_secret: your-client-secret
      token_url: https://auth.example.com/token

- secure_api:
    url: https://api.secure.com/openapi.json
    type: openapi
    auth:
      type: bearer
      value: Bearer ${SECURE_API_TOKEN}

# Set in .env file
export SECURE_API_TOKEN="your-secure-token-here"

# List registered APIs
cuga apis list

# Show API details
cuga apis info myapi

# Test API connectivity
cuga apis test myapi

# Refresh API schemas
cuga apis refresh

# Remove API
cuga apis remove myapi

# Check API health
cuga apis health

# Monitor API performance
cuga apis monitor

# View API logs
cuga apis logs myapi

# Test basic connectivity
cuga apis test myapi

# Test with authentication
cuga apis test myapi --auth

# Test specific endpoints
cuga apis test myapi --endpoint /contacts

# Start CUGA with your API
cuga start demo

# Test natural language queries
"get contacts from my custom API"
"analyze data from the integrated service"

# Load test your API
cuga apis load-test myapi --requests 100

# Stress test
cuga apis stress-test myapi --duration 300

# Check API registration
cuga apis list --verbose

# Test API connectivity
cuga apis test myapi --debug

# View API logs
cuga apis logs myapi --tail

# Check configuration
cuga config show --section apis

# Validate OpenAPI spec
curl -s https://api.example.com/openapi.json | jq .

# Test API endpoint directly
curl -H "Authorization: Bearer token" https://api.example.com/endpoint

# Check CUGA logs
cuga logs --service api-registry