MCP server for set_avatar

README.md

MCP server for update_avatar

pyproject.toml

[project]
name = "update_avatar.py"
version = "1.1.1"
description = "MCP server for update_avatar"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "fastmcp>=2.13.1",
]

update_avatar.py

#!/usr/bin/env python3
"""
MCP Server for Avatar Emotion and Animation Control

This server provides tools to control an avatar's facial expressions and body animations
through a pubsub messaging system.
"""
import os, time, json
from typing import Optional
from fastmcp import FastMCP
from traceback import print_exc

# Initialize MCP server
mcp = FastMCP('avatar-control')

WS_URL = os.getenv('WS_URL', 'https://ai.ccl.io/ws/')

# Valid expressions and animations
VALID_EXPRESSIONS = 'happy sad angry surprised confused neutral excited'.split()
VALID_ANIMATIONS =  'dance-rumba wave nod jump sit stand walk run'.split()


class PubSubClient:
    """Placeholder for your custom pubsub client"""
    def __init__(self, url: str) -> None:
        self.url = url
        self.ws = None
        self.connected = False
        self.reconnect_thread = None
        self.should_reconnect = True
        self.max_retries = 5
        self.channel = 'sup-in'
        self.connect()
        return
    def _connect(self) -> None:
        self.ws = websocket.create_connection(self.url, timeout=10)
        self.connected = True
        return
    def _close(self) -> None:
        """Close websocket connection"""
        try:
            if self.ws:
                self.ws.close()
        except:
            pass
        self.connected = False
        return
    def connect(self) -> None:
        """Establish websocket connection"""
        self._close()
        self._connect()
        print(f"Connected to pubsub server at {self.url}")
        return
    def _send(self, channel: str, message: str) -> None:
        self.ws.send(channel)
        self.ws.send(message)
        return
    def _publish(self, channel: str, params: dict) -> None:
        """Publish a message to the pubsub server"""
        msg = json.dumps(dict(
            method='pub',
            params=params,
        ))
        for retry in range(1, self.max_retries):
            try   : self._send(channel, msg)
            except: pass
            pass
        self._send(self.channel, msg)
        return
    def publish(self, expression: str, animation: str) -> None:
        params = {'from': 'update-avatar'}
        if expression: params['expression'] = expression
        if animation : params[ 'animation'] = animation
        self._publish(channel, params)
        return
    pass


# Initialize pubsub client
pubsub_client = PubSubClient(WS_URL)


@mcp.tool()
def update_avatar(
        expression: Optional[str] = None,
        animation: Optional[str] = None
) -> str:
    """Update avatar facial expression and/or body animation.

    Args:
    	expression: happy | sad | angry | surprised | confused | neutral | excited
    	animation: dance-rumba | wave | nod | jump | sit | stand | walk | run

    Both optional. Returns JSON.
    """
    # Normalize falsy values to ''
    expression = expression if expression else ''
    animation  =  animation if  animation else ''
    # Validate expression
    if expression and expression not in VALID_EXPRESSIONS:
        return json.dumps({
            "status": "error",
            "error": "invalid_expression",
            "message": f"'{expression}' is not valid. Valid: {', '.join(VALID_EXPRESSIONS)}"
        })
    # Validate animation
    if animation and animation not in VALID_ANIMATIONS:
        return json.dumps({
            "status": "error",
            "error": "invalid_animation",
            "message": f"'{animation}' is not valid. Valid: {', '.join(VALID_ANIMATIONS)}"
        })
    # Check if any changes requested
    if not expression and not animation:
        return json.dumps({
            "status": "error",
            "error": "no_changes",
            "message": "No changes requested. Provide expression and/or animation."
        })
    # Publish to pubsub
    try:
        pubsub_client.publish("avatar/control", {
            "expression": expression,
            "animation": animation
        })
        return json.dumps({
            "status": "success",
            "expression": expression,
            "animation": animation
        })
    except Exception as e:
        return json.dumps({
            "status": "error",
            "error": "publish_failed",
            "message": str(e)
        })    
    pass


if __name__ == "__main__":
    mcp.run()