Chat Tools

sequenceDiagram
    participant U as User
    participant O as Omi AI
    participant Y as Your Server

    U->>O: "Send hello to #general"
    O->>O: AI decides to use your tool
    O->>Y: POST /api/send_message
    Note over Y: {uid, app_id, tool_name, channel, message}
    Y->>Y: Process request
    Y-->>O: {result: "Message sent!"}
    O-->>U: "Done! Message sent to #general"

- Accept POST requests with JSON payload (or GET with query parameters)
- Receive these standard fields:
  - `uid`: User ID
  - `app_id`: Your app ID
  - `tool_name`: Name of the tool being called
  - Plus any tool-specific parameters
- Return JSON with either:
  - `result`: Success message (string)
  - `error`: Error message (string)

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/api/send_message', methods=['POST'])
def send_message():
    """
    Tool endpoint: Send a message to a Slack channel

    Expected payload:
    {
        "uid": "user_id",
        "app_id": "slack_app_id",
        "tool_name": "send_slack_message",
        "channel": "#general",
        "message": "Hello from Omi!"
    }
    """
    data = request.json

    # Validate required parameters
    if not data:
        return jsonify({'error': 'Missing request body'}), 400

    uid = data.get('uid')
    channel = data.get('channel')
    message = data.get('message')

    if not uid:
        return jsonify({'error': 'Missing uid parameter'}), 400
    if not channel:
        return jsonify({'error': 'Missing required parameter: channel'}), 400
    if not message:
        return jsonify({'error': 'Missing required parameter: message'}), 400

    # Get user's authentication token (from your database)
    slack_token = get_user_token(uid)
    if not slack_token:
        return jsonify({
            'error': 'Slack not connected. Please connect your Slack account.'
        }), 401

    # Call external API (e.g., Slack API)
    slack_response = requests.post(
        'https://slack.com/api/chat.postMessage',
        headers={'Authorization': f'Bearer {slack_token}'},
        json={'channel': channel, 'text': message}
    )

    if slack_response.json().get('ok'):
        return jsonify({
            'result': f'Successfully sent message to {channel}'
        })
    else:
        return jsonify({
            'error': f"Failed to send message: {slack_response.json().get('error')}"
        }), 400

@app.route('/api/search_messages', methods=['POST'])
def search_messages():
    """
    Tool endpoint: Search for messages

    Expected payload:
    {
        "uid": "user_id",
        "app_id": "slack_app_id",
        "tool_name": "search_slack_messages",
        "query": "meeting notes",
        "channel": "#general"  # optional
    }
    """
    data = request.json

    uid = data.get('uid')
    query = data.get('query')
    channel = data.get('channel')

    if not uid:
        return jsonify({'error': 'Missing uid parameter'}), 400
    if not query:
        return jsonify({'error': 'Missing required parameter: query'}), 400

    # Get user's authentication token
    slack_token = get_user_token(uid)
    if not slack_token:
        return jsonify({
            'error': 'Slack not connected. Please connect your Slack account.'
        }), 401

    # Build search query
    search_query = query
    if channel:
        search_query = f'in:{channel} {query}'

    # Call external API
    slack_response = requests.get(
        'https://slack.com/api/search.messages',
        headers={'Authorization': f'Bearer {slack_token}'},
        params={'query': search_query}
    )

    if slack_response.json().get('ok'):
        messages = slack_response.json().get('messages', {}).get('matches', [])
        if not messages:
            return jsonify({
                'result': f'No messages found for "{query}"'
            })

        results = []
        for msg in messages[:5]:
            results.append(f"- {msg.get('text', '')[:100]} (in #{msg.get('channel', {}).get('name', 'unknown')})")

        return jsonify({
            'result': f'Found {len(messages)} messages:\n' + '\n'.join(results)
        })
    else:
        return jsonify({
            'error': f"Failed to search messages: {slack_response.json().get('error')}"
        }), 400

def get_user_token(uid: str) -> Optional[str]:
    """Get user's authentication token from database"""
    # In production, fetch from secure database
    return user_tokens.get(uid)

@app.route('/api/send_message', methods=['POST'])
def send_message():
    data = request.json
    uid = data.get('uid')

    # Check authentication
    token = get_user_token(uid)
    if not token:
        return jsonify({
            'error': 'Account not connected. Please connect your account in app settings.'
        }), 401

    # Proceed with tool execution
    # ...

# Good error messages
return jsonify({'error': 'Slack not connected. Please connect your Slack account.'}), 401
return jsonify({'error': 'Missing required parameter: channel'}), 400
return jsonify({'error': 'Channel not found. Please check the channel name.'}), 404

# Avoid exposing sensitive information
# ❌ Bad: return jsonify({'error': f'Database connection failed: {db_password}'}), 500
# ✅ Good: return jsonify({'error': 'Service temporarily unavailable. Please try again later.'}), 500

{
  "tools": [
    {
      "name": "send_slack_message",
      "description": "Send a message to a Slack channel. Use this when the user wants to send a message, post an update, or notify a channel in Slack.",
      "endpoint": "/api/send_message",
      "method": "POST",
      "parameters": {
        "properties": {
          "channel": {
            "type": "string",
            "description": "The Slack channel to send to (e.g., '#general')"
          },
          "message": {
            "type": "string",
            "description": "The message text to send"
          }
        },
        "required": ["channel", "message"]
      },
      "auth_required": true,
      "status_message": "Sending message to Slack..."
    },
    {
      "name": "search_slack_messages",
      "description": "Search for messages in Slack. Use this when the user wants to find specific messages or look up past conversations.",
      "endpoint": "/api/search_messages",
      "method": "POST",
      "parameters": {
        "properties": {
          "query": {
            "type": "string",
            "description": "Search query"
          },
          "channel": {
            "type": "string",
            "description": "Optional channel to search in"
          }
        },
        "required": ["query"]
      },
      "auth_required": true,
      "status_message": "Searching Slack..."
    }
  ]
}

{
  "parameters": {
    "properties": {
      "param_name": {
        "type": "string | integer | boolean | array | object",
        "description": "Description of what this parameter does"
      }
    },
    "required": ["param_name"]
  }
}

from fastapi import FastAPI

app = FastAPI()

@app.get("/.well-known/omi-tools.json")
async def get_omi_tools_manifest():
    return {
        "tools": [
            {
                "name": "search_songs",
                "description": "Search for songs on Spotify",
                "endpoint": "/tools/search_songs",
                "method": "POST",
                "parameters": {
                    "properties": {
                        "query": {"type": "string", "description": "Search query"},
                        "limit": {"type": "integer", "description": "Max results"}
                    },
                    "required": ["query"]
                },
                "auth_required": True,
                "status_message": "Searching Spotify..."
            },
            {
                "name": "add_to_playlist",
                "description": "Add a song to a playlist",
                "endpoint": "/tools/add_to_playlist",
                "method": "POST",
                "parameters": {
                    "properties": {
                        "song_name": {"type": "string", "description": "Song name"},
                        "artist_name": {"type": "string", "description": "Artist name"},
                        "playlist_name": {"type": "string", "description": "Playlist name"}
                    },
                    "required": ["song_name"]
                },
                "auth_required": True,
                "status_message": "Adding to playlist..."
            }
        ],
        "chat_messages": {
            "enabled": True,
            "target": "app",
            "notify": False
        }
    }

{
  "tools": [...],
  "chat_messages": {
    "enabled": true,
    "target": "app",
    "notify": false
  }
}

curl -X POST https://api.omi.me/v1/integrations/notification \
  -H "Authorization: Bearer YOUR_APP_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "uid": "user_firebase_id",
    "aid": "your_app_id",
    "message": "Your task has completed! Here are the results..."
  }'

import httpx

async def send_task_result(uid: str, result: str):
    """Send task completion message to user's chat."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.omi.me/v1/integrations/notification",
            headers={
                "Authorization": f"Bearer {APP_API_KEY}",
                "Content-Type": "application/json",
            },
            json={
                "uid": uid,
                "aid": APP_ID,
                "message": result,
            },
        )
        return response.status_code == 200

| Good | Bad |
|------|-----|
| `send_slack_message` | `slack1` |
| `list_slack_channels` | `do_stuff` |
| `search_slack_messages` | `msg` |

- **When to use**: "Use this when the user wants to..."
- **Required parameters**: List what's needed
- **What it does**: Clear explanation of the action

**Example:**
```
Send a message to a Slack channel. Use this when the user wants to
send a message, post an update, or notify a channel in Slack.
Required parameters: channel (e.g., '#general') and message (the text to send).
```

| Method | Use For |
|--------|---------|
| `POST` | Creating resources or sending data (most common) |
| `GET` | Retrieving data (parameters as query params) |
| `PUT/PATCH` | Updating resources |
| `DELETE` | Deleting resources |

- "Searching Slack..."
- "Sending message..."
- "Creating calendar event..."

If not provided, Omi generates a default based on the tool name.

{
  "uid": "user_firebase_id",
  "app_id": "your_app_id",
  "tool_name": "send_slack_message",
  "channel": "#general",
  "message": "Hello from Omi!"
}

GET /api/search?uid=user_id&app_id=app_id&tool_name=search_slack_messages&query=meeting

- ✅ "Successfully sent message to #general"
- ❌ "OK"

- ✅ "Slack not connected. Please connect your account."
- ❌ "Error 401"

from functools import wraps
from flask import request, jsonify
import time

request_counts = {}

def rate_limit(max_requests=100, window=60):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            uid = request.json.get('uid')
            key = f"{uid}_{f.__name__}"
            now = time.time()

            if key in request_counts:
                requests, first_request = request_counts[key]
                if now - first_request < window:
                    if requests >= max_requests:
                        return jsonify({'error': 'Rate limit exceeded'}), 429
                    request_counts[key] = (requests + 1, first_request)
                else:
                    request_counts[key] = (1, now)
            else:
                request_counts[key] = (1, now)

            return f(*args, **kwargs)
        return wrapper
    return decorator

@app.route('/api/send_message', methods=['POST'])
@rate_limit(max_requests=100, window=60)
def send_message():
    # ...

```bash
curl -X POST https://your-server.com/api/send_message \
  -H "Content-Type: application/json" \
  -d '{
    "uid": "test_user_id",
    "app_id": "slack-integration",
    "tool_name": "send_slack_message",
    "channel": "#general",
    "message": "Test message"
  }'
```