example.py•1.15 kB
#!/usr/bin/env python3
"""Example usage of celery-mcp library."""
from celery_mcp import CeleryMCP
def main():
"""Demonstrate basic usage of CeleryMCP."""
print("Celery MCP Library Examples\n")
# Initialize the connector
print("1. Initializing CeleryMCP connector...")
mcp = CeleryMCP(broker_url='redis://localhost:6379/0')
print("Connector initialized successfully!\n")
# Example: Send a task
print("2. Sending a sample task...")
try:
result = mcp.send_task('add', args=[4, 4])
print(f"Task sent successfully. Task ID: {result.id}")
# Wait for result
print(" Waiting for result...")
task_result = result.get(timeout=10)
print(f"Task result: {task_result}")
except Exception as e:
print(f" Error: {e}")
print(" Note: This example requires a running Celery worker and Redis.\n")
# Example: Register a task
print("3. Registering a sample task...")
def multiply(x, y):
return x * y
mcp.register_task('multiply', multiply)
print("Task 'multiply' registered successfully!\n")
if __name__ == "__main__":
main()