0
# Extensions & Utilities
1
2
Extension system, utilities, converters, tasks, and additional functionality for organizing and enhancing Discord bots.
3
4
## Extensions System
5
6
### Basic Extension
7
8
```python
9
from interactions import Extension, slash_command, SlashContext, listen, events
10
11
class BasicExtension(Extension):
12
"""A basic extension example"""
13
14
@slash_command(name="ext_hello", description="Hello from extension")
15
async def hello_command(self, ctx: SlashContext):
16
await ctx.send("Hello from extension!")
17
18
@listen(events.Ready)
19
async def on_ready(self, event: events.Ready):
20
print(f"Extension loaded! Bot: {self.bot.user}")
21
22
# Load the extension
23
bot.load_extension("extensions.basic") # Loads from extensions/basic.py
24
```
25
26
### Extension with Setup Function
27
28
```python
29
# extensions/moderation.py
30
from interactions import Extension, slash_command, SlashContext, Client
31
32
class ModerationExtension(Extension):
33
def __init__(self, bot: Client):
34
self.banned_words = ["spam", "inappropriate"]
35
36
@slash_command(name="ban", description="Ban a user")
37
async def ban_command(self, ctx: SlashContext, user: Member, reason: str = None):
38
"""Ban a user from the guild"""
39
if not ctx.author.guild_permissions.BAN_MEMBERS:
40
await ctx.send("You don't have permission to ban members!", ephemeral=True)
41
return
42
43
await user.ban(reason=reason)
44
await ctx.send(f"Banned {user.mention}. Reason: {reason or 'No reason provided'}")
45
46
@slash_command(name="kick", description="Kick a user")
47
async def kick_command(self, ctx: SlashContext, user: Member, reason: str = None):
48
"""Kick a user from the guild"""
49
if not ctx.author.guild_permissions.KICK_MEMBERS:
50
await ctx.send("You don't have permission to kick members!", ephemeral=True)
51
return
52
53
await user.kick(reason=reason)
54
await ctx.send(f"Kicked {user.mention}. Reason: {reason or 'No reason provided'}")
55
56
@listen(events.MessageCreate)
57
async def check_banned_words(self, event: events.MessageCreate):
58
"""Auto-moderation for banned words"""
59
message = event.message
60
61
if message.author.bot:
62
return
63
64
content_lower = message.content.lower()
65
for word in self.banned_words:
66
if word in content_lower:
67
await message.delete()
68
await message.author.timeout(duration=300, reason=f"Used banned word: {word}")
69
break
70
71
def setup(bot):
72
"""Setup function called when extension is loaded"""
73
ModerationExtension(bot)
74
```
75
76
### Extension Lifecycle
77
78
```python
79
class LifecycleExtension(Extension):
80
def __init__(self, bot: Client):
81
print("Extension initialized")
82
83
def drop(self):
84
"""Called when extension is unloaded"""
85
print("Extension unloaded - cleaning up resources")
86
# Clean up resources, close connections, etc.
87
88
@Extension.listener
89
async def on_extension_load(self, event: events.ExtensionLoad):
90
"""Called when ANY extension loads"""
91
print(f"Extension loaded: {event.extension.name}")
92
93
@Extension.listener
94
async def on_extension_unload(self, event: events.ExtensionUnload):
95
"""Called when ANY extension unloads"""
96
print(f"Extension unloaded: {event.extension.name}")
97
98
def setup(bot):
99
LifecycleExtension(bot)
100
```
101
102
### Loading Extensions
103
104
```python
105
# Load single extension
106
bot.load_extension("extensions.moderation")
107
108
# Load multiple extensions
109
extensions = [
110
"extensions.moderation",
111
"extensions.fun",
112
"extensions.admin",
113
"extensions.utility"
114
]
115
116
for ext in extensions:
117
try:
118
bot.load_extension(ext)
119
print(f"Loaded {ext}")
120
except Exception as e:
121
print(f"Failed to load {ext}: {e}")
122
123
# Reload extension (unload then load)
124
bot.reload_extension("extensions.moderation")
125
126
# Unload extension
127
bot.unload_extension("extensions.moderation")
128
```
129
130
## Tasks & Scheduling
131
132
### Basic Recurring Task
133
134
```python
135
from interactions import Task, IntervalTrigger, Extension
136
import asyncio
137
138
class TaskExtension(Extension):
139
def __init__(self, bot: Client):
140
# Create task that runs every 60 seconds
141
self.status_task = Task(
142
coro=self.update_status,
143
trigger=IntervalTrigger(seconds=60)
144
)
145
self.status_task.start()
146
147
async def update_status(self):
148
"""Update bot status every minute"""
149
guild_count = len(self.bot.guilds)
150
activity = Activity(
151
name=f"Watching {guild_count} servers",
152
type=ActivityType.WATCHING
153
)
154
await self.bot.change_presence(activity=activity)
155
156
def drop(self):
157
"""Clean up when extension unloads"""
158
self.status_task.stop()
159
160
def setup(bot):
161
TaskExtension(bot)
162
```
163
164
### Time-Based Tasks
165
166
```python
167
from interactions import Task, TimeTrigger, DateTrigger, OrTrigger
168
from datetime import time, date, datetime
169
170
class ScheduledTaskExtension(Extension):
171
def __init__(self, bot: Client):
172
# Daily reminder at 9:00 AM
173
self.daily_reminder = Task(
174
coro=self.send_daily_reminder,
175
trigger=TimeTrigger(hour=9, minute=0)
176
)
177
178
# Specific date task
179
self.special_event = Task(
180
coro=self.special_event_reminder,
181
trigger=DateTrigger(year=2024, month=12, day=25, hour=12)
182
)
183
184
# Multiple time triggers
185
self.hourly_check = Task(
186
coro=self.hourly_maintenance,
187
trigger=OrTrigger(
188
TimeTrigger(minute=0), # Every hour at :00
189
TimeTrigger(minute=30) # Every hour at :30
190
)
191
)
192
193
# Start all tasks
194
self.daily_reminder.start()
195
self.special_event.start()
196
self.hourly_check.start()
197
198
async def send_daily_reminder(self):
199
"""Send daily reminder to all guilds"""
200
for guild in self.bot.guilds:
201
if guild.system_channel:
202
await guild.system_channel.send("π Daily reminder: Don't forget to check the announcements!")
203
204
async def special_event_reminder(self):
205
"""Special event reminder"""
206
for guild in self.bot.guilds:
207
if guild.system_channel:
208
await guild.system_channel.send("π Special event today!")
209
210
async def hourly_maintenance(self):
211
"""Perform maintenance tasks"""
212
print("Running hourly maintenance...")
213
# Clean up expired data, update caches, etc.
214
215
def setup(bot):
216
ScheduledTaskExtension(bot)
217
```
218
219
### Task Decorators
220
221
```python
222
from interactions import Task, IntervalTrigger
223
224
class DecoratorTaskExtension(Extension):
225
@Task.create(IntervalTrigger(minutes=5))
226
async def cleanup_task(self):
227
"""Cleanup task using decorator syntax"""
228
print("Running cleanup...")
229
# Cleanup logic here
230
231
@Task.create(IntervalTrigger(hours=1))
232
async def backup_data(self):
233
"""Hourly data backup"""
234
print("Backing up data...")
235
# Backup logic here
236
237
def setup(bot):
238
DecoratorTaskExtension(bot)
239
```
240
241
## Cooldowns & Rate Limiting
242
243
### Command Cooldowns
244
245
```python
246
from interactions import cooldown, Buckets, SlashContext
247
248
class CooldownExtension(Extension):
249
@slash_command(name="limited", description="Command with cooldown")
250
@cooldown(bucket=Buckets.USER, rate=1, per=30) # 1 use per 30 seconds per user
251
async def limited_command(self, ctx: SlashContext):
252
await ctx.send("This command has a 30-second cooldown per user!")
253
254
@slash_command(name="guild_limited", description="Guild-wide cooldown")
255
@cooldown(bucket=Buckets.GUILD, rate=3, per=60) # 3 uses per minute per guild
256
async def guild_limited_command(self, ctx: SlashContext):
257
await ctx.send("This command has a guild-wide cooldown!")
258
259
@slash_command(name="channel_limited", description="Channel cooldown")
260
@cooldown(bucket=Buckets.CHANNEL, rate=2, per=45) # 2 uses per 45 seconds per channel
261
async def channel_limited_command(self, ctx: SlashContext):
262
await ctx.send("This command has a per-channel cooldown!")
263
264
def setup(bot):
265
CooldownExtension(bot)
266
```
267
268
### Advanced Cooldown Systems
269
270
```python
271
from interactions import CooldownSystem, MaxConcurrency
272
273
class AdvancedCooldownExtension(Extension):
274
@slash_command(name="complex", description="Complex rate limiting")
275
@cooldown(bucket=Buckets.USER, rate=5, per=300) # 5 uses per 5 minutes
276
@cooldown(bucket=Buckets.GUILD, rate=20, per=300) # 20 uses per 5 minutes per guild
277
@max_concurrency(limit=1, per=Buckets.USER) # Only 1 concurrent use per user
278
async def complex_limited_command(self, ctx: SlashContext):
279
await ctx.defer() # This might take a while
280
281
# Simulate long operation
282
await asyncio.sleep(10)
283
284
await ctx.send("Complex operation completed!")
285
286
def setup(bot):
287
AdvancedCooldownExtension(bot)
288
```
289
290
### Custom Cooldown Handling
291
292
```python
293
@listen(events.CommandError)
294
async def cooldown_error_handler(event: events.CommandError):
295
"""Handle cooldown errors"""
296
if isinstance(event.error, interactions.errors.CommandOnCooldown):
297
ctx = event.ctx
298
error = event.error
299
300
retry_after = round(error.retry_after, 2)
301
await ctx.send(
302
f"Command is on cooldown! Try again in {retry_after} seconds.",
303
ephemeral=True
304
)
305
```
306
307
## Checks & Permissions
308
309
### Basic Checks
310
311
```python
312
from interactions import check, guild_only, dm_only, has_role, has_any_role, is_owner
313
314
class CheckExtension(Extension):
315
@slash_command(name="admin_only", description="Admin only command")
316
@check(lambda ctx: ctx.author.guild_permissions.ADMINISTRATOR)
317
async def admin_only_command(self, ctx: SlashContext):
318
await ctx.send("You are an administrator!")
319
320
@slash_command(name="guild_only", description="Guild only command")
321
@guild_only()
322
async def guild_only_command(self, ctx: SlashContext):
323
await ctx.send("This only works in servers!")
324
325
@slash_command(name="dm_only", description="DM only command")
326
@dm_only()
327
async def dm_only_command(self, ctx: SlashContext):
328
await ctx.send("This only works in DMs!")
329
330
@slash_command(name="role_check", description="Requires specific role")
331
@has_role("Moderator")
332
async def role_check_command(self, ctx: SlashContext):
333
await ctx.send("You have the Moderator role!")
334
335
@slash_command(name="any_role_check", description="Requires one of several roles")
336
@has_any_role("Admin", "Moderator", "Helper")
337
async def any_role_check_command(self, ctx: SlashContext):
338
await ctx.send("You have at least one of the required roles!")
339
340
@slash_command(name="owner_only", description="Bot owner only")
341
@is_owner()
342
async def owner_only_command(self, ctx: SlashContext):
343
await ctx.send("You are the bot owner!")
344
345
def setup(bot):
346
CheckExtension(bot)
347
```
348
349
### Custom Checks
350
351
```python
352
def is_premium_user():
353
"""Custom check for premium users"""
354
async def predicate(ctx: SlashContext):
355
# Check if user is premium (from database, etc.)
356
return await check_premium_status(ctx.author.id)
357
return check(predicate)
358
359
def in_allowed_channel(*channel_ids):
360
"""Check if command is used in allowed channels"""
361
async def predicate(ctx: SlashContext):
362
return ctx.channel.id in channel_ids
363
return check(predicate)
364
365
class CustomCheckExtension(Extension):
366
@slash_command(name="premium", description="Premium users only")
367
@is_premium_user()
368
async def premium_command(self, ctx: SlashContext):
369
await ctx.send("Welcome, premium user! π")
370
371
@slash_command(name="channel_restricted", description="Only in specific channels")
372
@in_allowed_channel(123456789, 987654321) # Replace with actual channel IDs
373
async def channel_restricted_command(self, ctx: SlashContext):
374
await ctx.send("This command works in this channel!")
375
376
def setup(bot):
377
CustomCheckExtension(bot)
378
```
379
380
## Converters
381
382
### Built-in Converters
383
384
The library provides many built-in converters for automatic argument conversion:
385
386
```python
387
from interactions import (
388
UserConverter, MemberConverter, RoleConverter, GuildConverter,
389
ChannelConverter, MessageConverter, CustomEmojiConverter,
390
SnowflakeConverter, Converter
391
)
392
393
# Converters are used automatically based on parameter types
394
@slash_command(name="info", description="Get info about various objects")
395
async def info_command(
396
ctx: SlashContext,
397
user: User, # Automatically converted using UserConverter
398
channel: GuildText, # Automatically converted using ChannelConverter
399
role: Role # Automatically converted using RoleConverter
400
):
401
await ctx.send(f"User: {user}, Channel: {channel.mention}, Role: {role.mention}")
402
```
403
404
### Custom Converter
405
406
```python
407
from interactions import Converter
408
409
class TemperatureConverter(Converter):
410
"""Convert temperature strings to celsius"""
411
412
async def convert(self, ctx: SlashContext, argument: str) -> float:
413
"""Convert temperature to celsius"""
414
argument = argument.lower().strip()
415
416
if argument.endswith('f') or argument.endswith('Β°f'):
417
# Convert Fahrenheit to Celsius
418
temp = float(argument.rstrip('fΒ°'))
419
return (temp - 32) * 5/9
420
421
elif argument.endswith('k'):
422
# Convert Kelvin to Celsius
423
temp = float(argument.rstrip('k'))
424
return temp - 273.15
425
426
elif argument.endswith('c') or argument.endswith('Β°c'):
427
# Already Celsius
428
return float(argument.rstrip('cΒ°'))
429
430
else:
431
# Assume Celsius if no unit
432
try:
433
return float(argument)
434
except ValueError:
435
raise ValueError(f"Invalid temperature format: {argument}")
436
437
# Use custom converter
438
@slash_command(name="temperature", description="Convert temperature")
439
async def temperature_command(ctx: SlashContext, temp: TemperatureConverter):
440
"""Command using custom temperature converter"""
441
fahrenheit = (temp * 9/5) + 32
442
kelvin = temp + 273.15
443
444
await ctx.send(f"{temp}Β°C = {fahrenheit}Β°F = {kelvin}K")
445
```
446
447
## Wait/Synchronization
448
449
### Wait for Events
450
451
```python
452
from interactions import Wait
453
454
class WaitExtension(Extension):
455
@slash_command(name="wait_demo", description="Demonstrate wait functionality")
456
async def wait_demo_command(self, ctx: SlashContext):
457
await ctx.send("Say something in the next 30 seconds...")
458
459
try:
460
# Wait for a message from the same user in the same channel
461
message_event = await Wait.for_message(
462
self.bot,
463
check=lambda m: m.author.id == ctx.author.id and m.channel.id == ctx.channel.id,
464
timeout=30
465
)
466
467
message = message_event.message
468
await ctx.followup(f"You said: {message.content}")
469
470
except asyncio.TimeoutError:
471
await ctx.followup("You didn't say anything in time!")
472
473
@slash_command(name="wait_reaction", description="Wait for reaction")
474
async def wait_reaction_command(self, ctx: SlashContext):
475
msg = await ctx.send("React with π or π")
476
477
# Add reactions
478
await msg.add_reaction("π")
479
await msg.add_reaction("π")
480
481
try:
482
# Wait for reaction from command user
483
reaction_event = await Wait.for_reaction(
484
self.bot,
485
message=msg,
486
check=lambda r, u: u.id == ctx.author.id and str(r.emoji) in ["π", "π"],
487
timeout=60
488
)
489
490
reaction, user = reaction_event.reaction, reaction_event.user
491
if str(reaction.emoji) == "π":
492
await ctx.followup("You liked it!")
493
else:
494
await ctx.followup("You didn't like it.")
495
496
except asyncio.TimeoutError:
497
await ctx.followup("No reaction received!")
498
499
def setup(bot):
500
WaitExtension(bot)
501
```
502
503
### Wait for Components
504
505
```python
506
@slash_command(name="wait_button", description="Wait for button press")
507
async def wait_button_command(self, ctx: SlashContext):
508
button = Button(
509
style=ButtonStyle.PRIMARY,
510
label="Click Me!",
511
custom_id="wait_button"
512
)
513
514
await ctx.send("Click the button:", components=[ActionRow(button)])
515
516
try:
517
# Wait for component interaction
518
component_event = await Wait.for_component(
519
self.bot,
520
custom_id="wait_button",
521
timeout=30
522
)
523
524
component_ctx = component_event.ctx
525
await component_ctx.send("Button clicked!", ephemeral=True)
526
527
except asyncio.TimeoutError:
528
# Remove button after timeout
529
await ctx.edit_origin("Button expired.", components=[])
530
```
531
532
## Auto-Defer
533
534
### Automatic Response Deferral
535
536
```python
537
from interactions import auto_defer, AutoDefer
538
539
class AutoDeferExtension(Extension):
540
@slash_command(name="slow_command", description="A slow command")
541
@auto_defer() # Automatically defer after 2 seconds
542
async def slow_command(self, ctx: SlashContext):
543
# Simulate slow operation
544
await asyncio.sleep(5)
545
await ctx.send("Slow operation completed!")
546
547
@slash_command(name="custom_defer", description="Custom defer timing")
548
@auto_defer(time_until_defer=1.0, ephemeral=True) # Defer after 1 second, ephemeral
549
async def custom_defer_command(self, ctx: SlashContext):
550
await asyncio.sleep(3)
551
await ctx.send("Custom defer completed!")
552
553
def setup(bot):
554
AutoDeferExtension(bot)
555
```
556
557
### Global Auto-Defer
558
559
```python
560
# Configure auto-defer globally when creating the client
561
bot = Client(
562
token="TOKEN",
563
auto_defer=AutoDefer(enabled=True, time_until_defer=2.0, ephemeral=False)
564
)
565
```
566
567
## Utility Functions & Classes
568
569
### Greedy Argument Parsing
570
571
```python
572
from interactions import Greedy
573
574
@slash_command(name="ban_multiple", description="Ban multiple users")
575
async def ban_multiple_command(ctx: SlashContext, users: Greedy[Member], reason: str = None):
576
"""Ban multiple users with greedy parsing"""
577
if not users:
578
await ctx.send("No users specified!")
579
return
580
581
banned_count = 0
582
for user in users:
583
try:
584
await user.ban(reason=reason)
585
banned_count += 1
586
except Exception as e:
587
print(f"Failed to ban {user}: {e}")
588
589
await ctx.send(f"Successfully banned {banned_count} users.")
590
```
591
592
### Async Iterator
593
594
```python
595
from interactions import AsyncIterator
596
597
class UtilityExtension(Extension):
598
@slash_command(name="list_messages", description="List recent messages")
599
async def list_messages_command(self, ctx: SlashContext, limit: int = 10):
600
"""List recent messages using async iterator"""
601
messages = []
602
603
# AsyncIterator for channel history
604
async for message in AsyncIterator(ctx.channel.history(limit=limit)):
605
messages.append(f"{message.author.display_name}: {message.content[:50]}")
606
607
if messages:
608
message_list = "\n".join(messages[:10]) # Limit display
609
await ctx.send(f"Recent messages:\n```\n{message_list}\n```")
610
else:
611
await ctx.send("No messages found.")
612
613
def setup(bot):
614
UtilityExtension(bot)
615
```
616
617
### Typing Indicator
618
619
```python
620
from interactions import Typing
621
622
class TypingExtension(Extension):
623
@slash_command(name="typing_demo", description="Show typing indicator")
624
async def typing_demo_command(self, ctx: SlashContext):
625
await ctx.defer()
626
627
# Show typing indicator while processing
628
async with Typing(ctx.channel):
629
# Simulate work
630
await asyncio.sleep(3)
631
632
await ctx.send("Done processing!")
633
634
def setup(bot):
635
TypingExtension(bot)
636
```
637
638
## Advanced Extension Patterns
639
640
### Extension Dependencies
641
642
```python
643
# extensions/database.py
644
class DatabaseExtension(Extension):
645
def __init__(self, bot: Client):
646
self.connection = None
647
648
async def connect_database(self):
649
"""Connect to database"""
650
# Database connection logic
651
pass
652
653
def drop(self):
654
"""Close database connection"""
655
if self.connection:
656
self.connection.close()
657
658
def setup(bot):
659
DatabaseExtension(bot)
660
661
# extensions/user_management.py
662
class UserManagementExtension(Extension):
663
def __init__(self, bot: Client):
664
# Get database extension
665
self.db_ext = bot.get_extension("DatabaseExtension")
666
if not self.db_ext:
667
raise Exception("DatabaseExtension is required!")
668
669
@slash_command(name="user_info", description="Get user info from database")
670
async def user_info_command(self, ctx: SlashContext, user: User):
671
# Use database extension
672
user_data = await self.db_ext.get_user_data(user.id)
673
await ctx.send(f"User data: {user_data}")
674
675
def setup(bot):
676
UserManagementExtension(bot)
677
```
678
679
### Extension Configuration
680
681
```python
682
import json
683
import os
684
685
class ConfigurableExtension(Extension):
686
def __init__(self, bot: Client):
687
# Load configuration
688
config_path = "extensions/config/moderation.json"
689
self.config = self.load_config(config_path)
690
691
# Use config values
692
self.auto_mod_enabled = self.config.get("auto_mod_enabled", True)
693
self.banned_words = self.config.get("banned_words", [])
694
self.warning_threshold = self.config.get("warning_threshold", 3)
695
696
def load_config(self, path: str) -> dict:
697
"""Load configuration from JSON file"""
698
if os.path.exists(path):
699
with open(path, 'r') as f:
700
return json.load(f)
701
return {}
702
703
def save_config(self, path: str):
704
"""Save configuration to JSON file"""
705
os.makedirs(os.path.dirname(path), exist_ok=True)
706
with open(path, 'w') as f:
707
json.dump(self.config, f, indent=4)
708
709
@slash_command(name="config", description="Configure extension")
710
async def config_command(self, ctx: SlashContext, setting: str, value: str):
711
"""Update configuration setting"""
712
if setting in self.config:
713
# Type conversion based on current value
714
current_value = self.config[setting]
715
if isinstance(current_value, bool):
716
self.config[setting] = value.lower() in ('true', '1', 'yes', 'on')
717
elif isinstance(current_value, int):
718
self.config[setting] = int(value)
719
elif isinstance(current_value, list):
720
self.config[setting] = value.split(',')
721
else:
722
self.config[setting] = value
723
724
self.save_config("extensions/config/moderation.json")
725
await ctx.send(f"Updated {setting} to {value}")
726
else:
727
await ctx.send(f"Unknown setting: {setting}")
728
729
def setup(bot):
730
ConfigurableExtension(bot)
731
```