0
# MCP Integration
1
2
Model Context Protocol (MCP) integration for building AI-enabled web applications with resource registration, tool definitions, and prompt templates that allow AI agents to interact with your Robyn application.
3
4
## Capabilities
5
6
### MCP Application Interface
7
8
The MCP interface is accessible through the `mcp` property of the Robyn application instance, providing methods to register resources, tools, and prompts for AI agent consumption.
9
10
```python { .api }
11
class MCPApp:
12
def resource(self, uri: str, name: str, description: str):
13
"""
14
Register an MCP resource that AI agents can access.
15
16
Args:
17
uri: Unique resource identifier (URI format)
18
name: Human-readable name for the resource
19
description: Description of what the resource provides
20
21
Usage:
22
@app.mcp.resource("users://all", "User List", "Complete list of system users")
23
def get_all_users():
24
return {"users": [...]}
25
"""
26
27
def tool(self, name: str, description: str, input_schema: dict):
28
"""
29
Register an MCP tool that AI agents can invoke.
30
31
Args:
32
name: Tool name (used by agents to invoke the tool)
33
description: Description of what the tool does
34
input_schema: JSON schema defining tool input parameters
35
36
Usage:
37
@app.mcp.tool(
38
"create_user",
39
"Create a new user account",
40
{
41
"type": "object",
42
"properties": {
43
"username": {"type": "string"},
44
"email": {"type": "string", "format": "email"}
45
},
46
"required": ["username", "email"]
47
}
48
)
49
def create_user_tool(username, email):
50
return {"user_id": 123, "message": "User created"}
51
"""
52
53
def prompt(self, name: str, description: str):
54
"""
55
Register an MCP prompt template for AI agents.
56
57
Args:
58
name: Prompt template name
59
description: Description of the prompt's purpose
60
61
Usage:
62
@app.mcp.prompt("user_summary", "Generate a summary of user activity")
63
def user_summary_prompt(user_id):
64
return f"Summarize activity for user {user_id}"
65
"""
66
```
67
68
## Usage Examples
69
70
### Basic MCP Resource Registration
71
72
```python
73
from robyn import Robyn
74
75
app = Robyn(__file__)
76
77
# Sample data
78
users_db = [
79
{"id": 1, "username": "alice", "email": "alice@example.com", "active": True},
80
{"id": 2, "username": "bob", "email": "bob@example.com", "active": False},
81
{"id": 3, "username": "charlie", "email": "charlie@example.com", "active": True}
82
]
83
84
# Register MCP resources that AI agents can access
85
@app.mcp.resource("users://all", "All Users", "Complete list of all users in the system")
86
def get_all_users():
87
"""Resource providing all users"""
88
return {"users": users_db, "count": len(users_db)}
89
90
@app.mcp.resource("users://active", "Active Users", "List of currently active users")
91
def get_active_users():
92
"""Resource providing only active users"""
93
active_users = [user for user in users_db if user["active"]]
94
return {"users": active_users, "count": len(active_users)}
95
96
@app.mcp.resource("system://stats", "System Statistics", "Current system statistics and metrics")
97
def get_system_stats():
98
"""Resource providing system statistics"""
99
return {
100
"total_users": len(users_db),
101
"active_users": len([u for u in users_db if u["active"]]),
102
"server_uptime": "5 days",
103
"memory_usage": "256 MB"
104
}
105
106
# Regular HTTP endpoints
107
@app.get("/users")
108
def list_users(request):
109
return {"users": users_db}
110
111
app.start()
112
```
113
114
### MCP Tools for AI Agent Actions
115
116
```python
117
from robyn import Robyn
118
import json
119
import uuid
120
121
app = Robyn(__file__)
122
123
# In-memory data store
124
users_db = {}
125
tasks_db = {}
126
127
# MCP Tools - AI agents can invoke these
128
@app.mcp.tool(
129
"create_user",
130
"Create a new user account with username and email",
131
{
132
"type": "object",
133
"properties": {
134
"username": {
135
"type": "string",
136
"description": "Unique username for the account",
137
"minLength": 3,
138
"maxLength": 50
139
},
140
"email": {
141
"type": "string",
142
"format": "email",
143
"description": "User's email address"
144
},
145
"full_name": {
146
"type": "string",
147
"description": "User's full name"
148
}
149
},
150
"required": ["username", "email"]
151
}
152
)
153
def create_user_tool(username, email, full_name=None):
154
"""Tool for creating users that AI agents can invoke"""
155
user_id = str(uuid.uuid4())
156
157
# Check if username already exists
158
if any(user["username"] == username for user in users_db.values()):
159
return {
160
"success": False,
161
"error": "Username already exists",
162
"user_id": None
163
}
164
165
new_user = {
166
"id": user_id,
167
"username": username,
168
"email": email,
169
"full_name": full_name,
170
"created_at": "2023-12-01T12:00:00Z",
171
"active": True
172
}
173
174
users_db[user_id] = new_user
175
176
return {
177
"success": True,
178
"message": "User created successfully",
179
"user_id": user_id,
180
"user": new_user
181
}
182
183
@app.mcp.tool(
184
"create_task",
185
"Create a new task for a user",
186
{
187
"type": "object",
188
"properties": {
189
"user_id": {
190
"type": "string",
191
"description": "ID of the user to assign the task to"
192
},
193
"title": {
194
"type": "string",
195
"description": "Task title",
196
"maxLength": 200
197
},
198
"description": {
199
"type": "string",
200
"description": "Detailed task description"
201
},
202
"priority": {
203
"type": "string",
204
"enum": ["low", "medium", "high"],
205
"description": "Task priority level"
206
}
207
},
208
"required": ["user_id", "title"]
209
}
210
)
211
def create_task_tool(user_id, title, description=None, priority="medium"):
212
"""Tool for creating tasks that AI agents can invoke"""
213
if user_id not in users_db:
214
return {
215
"success": False,
216
"error": "User not found",
217
"task_id": None
218
}
219
220
task_id = str(uuid.uuid4())
221
new_task = {
222
"id": task_id,
223
"user_id": user_id,
224
"title": title,
225
"description": description,
226
"priority": priority,
227
"status": "pending",
228
"created_at": "2023-12-01T12:00:00Z"
229
}
230
231
tasks_db[task_id] = new_task
232
233
return {
234
"success": True,
235
"message": "Task created successfully",
236
"task_id": task_id,
237
"task": new_task
238
}
239
240
@app.mcp.tool(
241
"search_users",
242
"Search for users by username or email",
243
{
244
"type": "object",
245
"properties": {
246
"query": {
247
"type": "string",
248
"description": "Search query to match against username or email",
249
"minLength": 1
250
},
251
"active_only": {
252
"type": "boolean",
253
"description": "Only return active users",
254
"default": False
255
}
256
},
257
"required": ["query"]
258
}
259
)
260
def search_users_tool(query, active_only=False):
261
"""Tool for searching users that AI agents can invoke"""
262
query_lower = query.lower()
263
results = []
264
265
for user in users_db.values():
266
if active_only and not user["active"]:
267
continue
268
269
if (query_lower in user["username"].lower() or
270
query_lower in user["email"].lower() or
271
(user["full_name"] and query_lower in user["full_name"].lower())):
272
results.append(user)
273
274
return {
275
"query": query,
276
"results": results,
277
"count": len(results),
278
"active_only": active_only
279
}
280
281
# Regular HTTP endpoints for web interface
282
@app.get("/users")
283
def list_users(request):
284
return {"users": list(users_db.values())}
285
286
@app.get("/tasks")
287
def list_tasks(request):
288
return {"tasks": list(tasks_db.values())}
289
290
app.start()
291
```
292
293
### MCP Prompt Templates
294
295
```python
296
from robyn import Robyn
297
import json
298
299
app = Robyn(__file__)
300
301
# Sample data
302
users_db = {
303
"1": {"username": "alice", "email": "alice@example.com", "role": "admin"},
304
"2": {"username": "bob", "email": "bob@example.com", "role": "user"}
305
}
306
307
projects_db = {
308
"proj1": {"name": "Website Redesign", "status": "active", "assigned_to": "1"},
309
"proj2": {"name": "Mobile App", "status": "completed", "assigned_to": "2"}
310
}
311
312
# MCP Prompt Templates - AI agents can use these for generating content
313
@app.mcp.prompt("user_report", "Generate a comprehensive report about a user")
314
def user_report_prompt(user_id):
315
"""Generate prompt for creating user reports"""
316
user = users_db.get(user_id)
317
if not user:
318
return "User not found"
319
320
user_projects = [p for p in projects_db.values() if p["assigned_to"] == user_id]
321
322
prompt = f"""
323
Generate a comprehensive report for the following user:
324
325
User Information:
326
- Username: {user['username']}
327
- Email: {user['email']}
328
- Role: {user['role']}
329
330
Assigned Projects: {len(user_projects)}
331
{json.dumps(user_projects, indent=2)}
332
333
Please include:
334
1. User activity summary
335
2. Project involvement analysis
336
3. Performance assessment
337
4. Recommendations for future assignments
338
"""
339
340
return prompt.strip()
341
342
@app.mcp.prompt("project_summary", "Generate a summary of project status and progress")
343
def project_summary_prompt(project_id=None):
344
"""Generate prompt for project summaries"""
345
if project_id:
346
project = projects_db.get(project_id)
347
if not project:
348
return "Project not found"
349
350
assigned_user = users_db.get(project["assigned_to"], {})
351
352
prompt = f"""
353
Generate a detailed summary for the following project:
354
355
Project: {project['name']}
356
Status: {project['status']}
357
Assigned to: {assigned_user.get('username', 'Unknown')} ({assigned_user.get('email', 'Unknown')})
358
359
Please analyze:
360
1. Current progress and milestones
361
2. Risk assessment
362
3. Resource allocation
363
4. Timeline adherence
364
5. Recommendations for improvement
365
"""
366
else:
367
prompt = f"""
368
Generate an executive summary of all projects:
369
370
Total Projects: {len(projects_db)}
371
372
Project Details:
373
{json.dumps(projects_db, indent=2)}
374
375
User Assignments:
376
{json.dumps(users_db, indent=2)}
377
378
Please provide:
379
1. Overall project portfolio health
380
2. Resource utilization analysis
381
3. Risk factors across projects
382
4. Strategic recommendations
383
"""
384
385
return prompt.strip()
386
387
@app.mcp.prompt("onboarding_guide", "Generate personalized onboarding guide for new users")
388
def onboarding_guide_prompt(user_role="user", department=None):
389
"""Generate onboarding prompts for new team members"""
390
prompt = f"""
391
Create a comprehensive onboarding guide for a new team member with the following profile:
392
393
Role: {user_role}
394
Department: {department or "General"}
395
396
Available System Features:
397
- User Management
398
- Project Tracking
399
- MCP AI Integration
400
- Real-time WebSocket Communication
401
402
Include:
403
1. Welcome message and company overview
404
2. Role-specific responsibilities and expectations
405
3. System access and navigation guide
406
4. Key contacts and resources
407
5. First-week checklist and goals
408
6. Training schedule and materials
409
"""
410
411
return prompt.strip()
412
413
# MCP Resource that provides prompt context
414
@app.mcp.resource("prompts://context", "Prompt Context Data", "Data context for prompt generation")
415
def get_prompt_context():
416
"""Provide context data for prompt generation"""
417
return {
418
"users": users_db,
419
"projects": projects_db,
420
"system_info": {
421
"total_users": len(users_db),
422
"active_projects": len([p for p in projects_db.values() if p["status"] == "active"]),
423
"features": ["User Management", "Project Tracking", "MCP Integration", "WebSocket Support"]
424
}
425
}
426
427
# Regular endpoints
428
@app.get("/")
429
def home(request):
430
return {
431
"message": "MCP-enabled Robyn application",
432
"features": {
433
"mcp_resources": 1,
434
"mcp_tools": 3,
435
"mcp_prompts": 3
436
}
437
}
438
439
app.start()
440
```
441
442
### Complete MCP Application Example
443
444
```python
445
from robyn import Robyn
446
import json
447
import uuid
448
from datetime import datetime, timedelta
449
450
app = Robyn(__file__)
451
452
# Data stores
453
users_db = {}
454
articles_db = {}
455
analytics_db = {"views": 0, "registrations": 0}
456
457
# === MCP RESOURCES ===
458
459
@app.mcp.resource("content://articles", "All Articles", "Complete article database")
460
def get_all_articles():
461
return {
462
"articles": list(articles_db.values()),
463
"count": len(articles_db),
464
"last_updated": datetime.now().isoformat()
465
}
466
467
@app.mcp.resource("analytics://dashboard", "Analytics Data", "Site analytics and metrics")
468
def get_analytics():
469
return {
470
"total_users": len(users_db),
471
"total_articles": len(articles_db),
472
"page_views": analytics_db["views"],
473
"registrations": analytics_db["registrations"],
474
"active_authors": len(set(a["author_id"] for a in articles_db.values()))
475
}
476
477
# === MCP TOOLS ===
478
479
@app.mcp.tool(
480
"publish_article",
481
"Publish a new article with title, content, and tags",
482
{
483
"type": "object",
484
"properties": {
485
"title": {"type": "string", "description": "Article title"},
486
"content": {"type": "string", "description": "Article content (markdown supported)"},
487
"author_id": {"type": "string", "description": "ID of the article author"},
488
"tags": {
489
"type": "array",
490
"items": {"type": "string"},
491
"description": "Article tags for categorization"
492
},
493
"publish_immediately": {
494
"type": "boolean",
495
"description": "Whether to publish immediately or save as draft",
496
"default": True
497
}
498
},
499
"required": ["title", "content", "author_id"]
500
}
501
)
502
def publish_article_tool(title, content, author_id, tags=None, publish_immediately=True):
503
if author_id not in users_db:
504
return {"success": False, "error": "Author not found"}
505
506
article_id = str(uuid.uuid4())
507
article = {
508
"id": article_id,
509
"title": title,
510
"content": content,
511
"author_id": author_id,
512
"author_name": users_db[author_id]["username"],
513
"tags": tags or [],
514
"status": "published" if publish_immediately else "draft",
515
"created_at": datetime.now().isoformat(),
516
"views": 0
517
}
518
519
articles_db[article_id] = article
520
521
return {
522
"success": True,
523
"message": f"Article {'published' if publish_immediately else 'saved as draft'} successfully",
524
"article_id": article_id,
525
"article": article
526
}
527
528
@app.mcp.tool(
529
"moderate_content",
530
"Review and moderate article content for policy compliance",
531
{
532
"type": "object",
533
"properties": {
534
"article_id": {"type": "string", "description": "ID of article to moderate"},
535
"action": {
536
"type": "string",
537
"enum": ["approve", "reject", "flag"],
538
"description": "Moderation action to take"
539
},
540
"reason": {"type": "string", "description": "Reason for the moderation action"}
541
},
542
"required": ["article_id", "action"]
543
}
544
)
545
def moderate_content_tool(article_id, action, reason=None):
546
if article_id not in articles_db:
547
return {"success": False, "error": "Article not found"}
548
549
article = articles_db[article_id]
550
551
if action == "approve":
552
article["status"] = "published"
553
article["moderated_at"] = datetime.now().isoformat()
554
elif action == "reject":
555
article["status"] = "rejected"
556
article["rejection_reason"] = reason
557
elif action == "flag":
558
article["status"] = "flagged"
559
article["flag_reason"] = reason
560
561
return {
562
"success": True,
563
"message": f"Article {action}ed successfully",
564
"article_id": article_id,
565
"new_status": article["status"]
566
}
567
568
# === MCP PROMPTS ===
569
570
@app.mcp.prompt("content_review", "Generate content review and improvement suggestions")
571
def content_review_prompt(article_id):
572
if article_id not in articles_db:
573
return "Article not found"
574
575
article = articles_db[article_id]
576
577
return f"""
578
Please review the following article and provide detailed feedback:
579
580
Title: {article['title']}
581
Author: {article['author_name']}
582
Tags: {', '.join(article['tags'])}
583
Word Count: {len(article['content'].split())}
584
585
Content:
586
{article['content']}
587
588
Please analyze:
589
1. Content quality and readability
590
2. SEO optimization opportunities
591
3. Factual accuracy and citations
592
4. Engagement potential
593
5. Specific improvement recommendations
594
"""
595
596
@app.mcp.prompt("author_performance", "Generate author performance analysis")
597
def author_performance_prompt(author_id):
598
if author_id not in users_db:
599
return "Author not found"
600
601
author = users_db[author_id]
602
author_articles = [a for a in articles_db.values() if a["author_id"] == author_id]
603
total_views = sum(a["views"] for a in author_articles)
604
605
return f"""
606
Generate a comprehensive performance analysis for author: {author['username']}
607
608
Author Statistics:
609
- Total Articles: {len(author_articles)}
610
- Total Views: {total_views}
611
- Average Views per Article: {total_views / len(author_articles) if author_articles else 0:.1f}
612
- Join Date: {author.get('created_at', 'Unknown')}
613
614
Article Breakdown:
615
{json.dumps([{
616
'title': a['title'],
617
'status': a['status'],
618
'views': a['views'],
619
'tags': a['tags']
620
} for a in author_articles], indent=2)}
621
622
Please provide:
623
1. Performance overview and trends
624
2. Content quality assessment
625
3. Engagement analysis
626
4. Areas for improvement
627
5. Recognition and achievements
628
"""
629
630
# Regular HTTP endpoints
631
@app.post("/register")
632
def register(request):
633
data = request.json()
634
user_id = str(uuid.uuid4())
635
636
users_db[user_id] = {
637
"id": user_id,
638
"username": data["username"],
639
"email": data["email"],
640
"created_at": datetime.now().isoformat()
641
}
642
643
analytics_db["registrations"] += 1
644
645
return {"message": "User registered", "user_id": user_id}
646
647
@app.get("/articles")
648
def list_articles(request):
649
analytics_db["views"] += 1
650
return {"articles": list(articles_db.values())}
651
652
@app.get("/articles/<article_id>")
653
def get_article(request):
654
article_id = request.path_params["article_id"]
655
if article_id in articles_db:
656
articles_db[article_id]["views"] += 1
657
analytics_db["views"] += 1
658
return {"article": articles_db[article_id]}
659
return Response(404, {}, {"error": "Article not found"})
660
661
app.start()
662
```