XSpoonAi Platform Integration
Connect SpoonOS agents to external platforms.
Integration Options
| Platform | Use Case | Transport |
|---|---|---|
| Telegram | Chat bot | Polling/Webhook |
| Discord | Server bot | Gateway/Webhook |
| Slack | Workspace bot | Events API |
| REST API | HTTP interface | FastAPI/Flask |
| Webhooks | Event triggers | HTTP POST |
| Scheduler | Cron tasks | APScheduler |
Telegram Bot
Setup
pip install python-telegram-bot
Implementation
# telegram_bot.py
import os
import asyncio
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters
from spoon_ai.agents import SpoonReactMCP
from spoon_ai.chat import ChatBot
from spoon_ai.tools import ToolManager
# Initialize agent
agent = SpoonReactMCP(
name="telegram_agent",
llm=ChatBot(model_name="gpt-4o"),
tools=ToolManager([]),
max_steps=10
)
async def start(update: Update, context):
"""Handle /start command."""
await update.message.reply_text(
"Hello! I'm your SpoonOS agent. Send me a message!"
)
async def handle_message(update: Update, context):
"""Process user messages through agent."""
user_message = update.message.text
user_id = update.effective_user.id
# Show typing indicator
await update.message.chat.send_action("typing")
try:
# Run agent
response = await agent.run(user_message)
# Split long messages (Telegram limit: 4096 chars)
if len(response) > 4000:
for i in range(0, len(response), 4000):
await update.message.reply_text(response[i:i+4000])
else:
await update.message.reply_text(response)
except Exception as e:
await update.message.reply_text(f"Error: {str(e)}")
def main():
"""Run the bot."""
token = os.getenv("TELEGRAM_BOT_TOKEN")
app = Application.builder().token(token).build()
app.add_handler(CommandHandler("start", start))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
print("Bot started...")
app.run_polling()
if __name__ == "__main__":
main()
Webhook Mode (Production)
# telegram_webhook.py
from fastapi import FastAPI, Request
from telegram import Update
from telegram.ext import Application
app = FastAPI()
telegram_app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
@app.post("/webhook")
async def webhook(request: Request):
data = await request.json()
update = Update.de_json(data, telegram_app.bot)
await telegram_app.process_update(update)
return {"ok": True}
# Set webhook:
# curl -X POST "https://api.telegram.org/bot<TOKEN>/setWebhook?url=https://your-domain.com/webhook"
Discord Bot
Setup
pip install discord.py
Implementation
# discord_bot.py
import os
import discord
from discord.ext import commands
from spoon_ai.agents import SpoonReactMCP
from spoon_ai.chat import ChatBot
from spoon_ai.tools import ToolManager
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)
# Initialize agent
agent = SpoonReactMCP(
name="discord_agent",
llm=ChatBot(model_name="gpt-4o"),
tools=ToolManager([]),
max_steps=10
)
@bot.event
async def on_ready():
print(f"Logged in as {bot.user}")
@bot.command(name="ask")
async def ask(ctx, *, question: str):
"""Ask the agent a question."""
async with ctx.typing():
try:
response = await agent.run(question)
# Split long messages (Discord limit: 2000 chars)
if len(response) > 1900:
for i in range(0, len(response), 1900):
await ctx.send(response[i:i+1900])
else:
await ctx.send(response)
except Exception as e:
await ctx.send(f"Error: {str(e)}")
@bot.event
async def on_message(message):
"""Handle mentions."""
if message.author == bot.user:
return
if bot.user.mentioned_in(message):
content = message.content.replace(f"<@{bot.user.id}>", "").strip()
if content:
async with message.channel.typing():
response = await agent.run(content)
await message.reply(response[:2000])
await bot.process_commands(message)
bot.run(os.getenv("DISCORD_BOT_TOKEN"))
Slash Commands
@bot.tree.command(name="analyze", description="Analyze a token")
async def analyze(interaction: discord.Interaction, token: str):
await interaction.response.defer()
response = await agent.run(f"Analyze token: {token}")
await interaction.followup.send(response[:2000])
# Sync commands on ready
@bot.event
async def on_ready():
await bot.tree.sync()
REST API Gateway
FastAPI Implementation
# api_gateway.py
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional
import asyncio
import uuid
from spoon_ai.agents import SpoonReactMCP
from spoon_ai.chat import ChatBot
from spoon_ai.tools import ToolManager
app = FastAPI(title="SpoonOS Agent API")
security = HTTPBearer()
# Agent pool for concurrent requests
agent_pool = {}
class QueryRequest(BaseModel):
query: str
session_id: Optional[str] = None
max_steps: Optional[int] = 10
class QueryResponse(BaseModel):
response: str
session_id: str
steps_taken: int
def get_or_create_agent(session_id: str) -> SpoonReactMCP:
"""Get existing agent or create new one."""
if session_id not in agent_pool:
agent_pool[session_id] = SpoonReactMCP(
name=f"api_agent_{session_id}",
llm=ChatBot(model_name="gpt-4o"),
tools=ToolManager([]),
max_steps=15
)
return agent_pool[session_id]
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Verify API token."""
if credentials.credentials != os.getenv("API_TOKEN"):
raise HTTPException(status_code=401, detail="Invalid token")
return credentials.credentials
@app.post("/v1/query", response_model=QueryResponse)
async def query_agent(
request: QueryRequest,
token: str = Depends(verify_token)
):
"""Execute agent query."""
session_id = request.session_id or str(uuid.uuid4())
agent = get_or_create_agent(session_id)
try:
response = await agent.run(request.query)
return QueryResponse(
response=response,
session_id=session_id,
steps_taken=agent.current_step
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/v1/query/stream")
async def query_agent_stream(request: QueryRequest):
"""Stream agent response."""
from fastapi.responses import StreamingResponse
session_id = request.session_id or str(uuid.uuid4())
agent = get_or_create_agent(session_id)
async def generate():
async for chunk in agent.stream(request.query):
yield f"data: {chunk}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
@app.delete("/v1/session/{session_id}")
async def delete_session(session_id: str):
"""Clean up session."""
if session_id in agent_pool:
del agent_pool[session_id]
return {"status": "deleted"}
# Run: uvicorn api_gateway:app --host 0.0.0.0 --port 8000
Webhook Handler
Generic Webhook Receiver
# webhook_handler.py
from fastapi import FastAPI, Request, BackgroundTasks
from pydantic import BaseModel
import hmac
import hashlib
app = FastAPI()
class WebhookConfig(BaseModel):
source: str
secret: str
agent_prompt_template: str
WEBHOOK_CONFIGS = {
"github": WebhookConfig(
source="github",
secret=os.getenv("GITHUB_WEBHOOK_SECRET"),
agent_prompt_template="Analyze this GitHub event: {event}"
),
"stripe": WebhookConfig(
source="stripe",
secret=os.getenv("STRIPE_WEBHOOK_SECRET"),
agent_prompt_template="Process this payment event: {event}"
)
}
def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
"""Verify webhook signature."""
expected = hmac.new(
secret.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
async def process_webhook(source: str, payload: dict):
"""Background task to process webhook."""
config = WEBHOOK_CONFIGS[source]
prompt = config.agent_prompt_template.format(event=payload)
agent = SpoonReactMCP(
name=f"webhook_{source}",
llm=ChatBot(model_name="gpt-4o"),
tools=ToolManager([])
)
result = await agent.run(prompt)
# Store result, send notification, etc.
print(f"Webhook processed: {result}")
@app.post("/webhook/{source}")
async def receive_webhook(
source: str,
request: Request,
background_tasks: BackgroundTasks
):
"""Receive and process webhooks."""
if source not in WEBHOOK_CONFIGS:
return {"error": "Unknown source"}, 400
body = await request.body()
signature = request.headers.get("X-Hub-Signature-256", "")
if not verify_signature(body, signature, WEBHOOK_CONFIGS[source].secret):
return {"error": "Invalid signature"}, 401
payload = await request.json()
background_tasks.add_task(process_webhook, source, payload)
return {"status": "accepted"}
Scheduled Tasks
APScheduler Integration
# scheduler.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
import asyncio
from spoon_ai.agents import SpoonReactMCP
from spoon_ai.chat import ChatBot
from spoon_ai.tools import ToolManager
scheduler = AsyncIOScheduler()
# Initialize agent
agent = SpoonReactMCP(
name="scheduled_agent",
llm=ChatBot(model_name="gpt-4o"),
tools=ToolManager([])
)
async def daily_market_report():
"""Generate daily market report."""
result = await agent.run(
"Generate a daily market report for top 10 cryptocurrencies"
)
# Send to Telegram, Discord, email, etc.
print(f"Daily report: {result}")
async def hourly_price_check():
"""Check prices hourly."""
result = await agent.run(
"Check current BTC and ETH prices, alert if >5% change"
)
print(f"Price check: {result}")
async def weekly_portfolio_review():
"""Weekly portfolio analysis."""
result = await agent.run(
"Review portfolio performance for the past week"
)
print(f"Weekly review: {result}")
# Schedule tasks
scheduler.add_job(
daily_market_report,
CronTrigger(hour=9, minute=0), # 9:00 AM daily
id="daily_report"
)
scheduler.add_job(
hourly_price_check,
CronTrigger(minute=0), # Every hour
id="hourly_check"
)
scheduler.add_job(
weekly_portfolio_review,
CronTrigger(day_of_week="mon", hour=10), # Monday 10 AM
id="weekly_review"
)
async def main():
scheduler.start()
print("Scheduler started...")
# Keep running
while True:
await asyncio.sleep(1)
if __name__ == "__main__":
asyncio.run(main())
Environment Configuration
# Platform tokens
TELEGRAM_BOT_TOKEN=...
DISCORD_BOT_TOKEN=...
SLACK_BOT_TOKEN=...
# API security
API_TOKEN=...
# Webhook secrets
GITHUB_WEBHOOK_SECRET=...
STRIPE_WEBHOOK_SECRET=...
# LLM
OPENAI_API_KEY=...
Best Practices
- Rate Limiting - Implement per-user limits
- Error Handling - Graceful degradation
- Logging - Structured logs for debugging
- Timeouts - Set appropriate timeouts
- Session Management - Clean up idle sessions