Compare commits

...
12 Commits
34 changed files with 4740 additions and 39 deletions

No files matched your search

+68
View File
@@ -0,0 +1,68 @@
name: Quick Debug Tests
on:
push:
branches: [ main ]
workflow_dispatch:
jobs:
debug-tests:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: 3.11
- name: Install UV
run: |
curl -LsSf https://astral.sh/uv/install.sh | sh
echo "$HOME/.local/bin" >> $GITHUB_PATH
- name: Install dependencies
run: |
cd src/praisonai
uv pip install --system ."[ui,gradio,api,agentops,google,openai,anthropic,cohere,chat,code,realtime,call,crewai,autogen]"
uv pip install --system duckduckgo_search
uv pip install --system pytest pytest-cov pytest-asyncio pytest-timeout
uv pip install --system "praisonaiagents[knowledge]"
- name: Set environment variables
run: |
echo "OPENAI_API_KEY=sk-test-key-for-github-actions-testing-only-not-real" >> $GITHUB_ENV
echo "OPENAI_API_BASE=https://api.openai.com/v1" >> $GITHUB_ENV
echo "OPENAI_MODEL_NAME=gpt-4o-mini" >> $GITHUB_ENV
echo "LOGLEVEL=WARNING" >> $GITHUB_ENV
echo "PYTHONPATH=${{ github.workspace }}/src/praisonai-agents:$PYTHONPATH" >> $GITHUB_ENV
- name: Run Unit Tests
id: pytest
run: |
cd src/praisonai
# Run pytest and save output, don't fail immediately
python -m pytest tests/unit/ -v --tb=short > test_output.txt 2>&1 || echo "TESTS_FAILED=true" >> $GITHUB_ENV
- name: Show Test Results
if: always()
run: |
cd src/praisonai
echo "=========================================="
echo "=== FULL TEST OUTPUT ==="
echo "=========================================="
cat test_output.txt
echo ""
echo "=========================================="
echo "=== FAILED TESTS SUMMARY ==="
echo "=========================================="
grep "^FAILED" test_output.txt || echo "No FAILED lines"
echo ""
# Fail if tests failed
if [ "$TESTS_FAILED" = "true" ]; then
echo "Tests failed!"
exit 1
fi
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.7.6" \
"praisonai>=2.7.7" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.7.6" \
"praisonai>=2.7.7" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.7.6" \
"praisonai>=2.7.7" \
"praisonai[ui]" \
"praisonai[crewai]"
+163
View File
@@ -0,0 +1,163 @@
# PraisonAI MCP Server Examples
This directory contains examples for running PraisonAI as an MCP (Model Context Protocol) server.
## Overview
PraisonAI can expose its capabilities via MCP protocol, allowing integration with:
- Claude Desktop
- Cursor
- Windsurf
- VSCode MCP clients
- Any MCP-compatible client
## Protocol Version
These examples use MCP Protocol Version **2025-11-25**.
## Examples
### 1. STDIO Server (stdio_server.py)
Run PraisonAI as an MCP server using STDIO transport (for Claude Desktop integration).
```bash
# Run directly
python stdio_server.py
# Or via CLI
praisonai mcp serve --transport stdio
```
### 2. HTTP Stream Server (http_stream_server.py)
Run PraisonAI as an MCP server using HTTP Stream transport.
```bash
# Run directly
python http_stream_server.py
# Or via CLI
praisonai mcp serve --transport http-stream --port 8080
```
### 3. Custom Tools Server (custom_tools_server.py)
Register custom tools and expose them via MCP.
```bash
python custom_tools_server.py
```
### 4. Client Example (mcp_client_example.py)
Example of connecting to a PraisonAI MCP server as a client.
```bash
python mcp_client_example.py
```
## CLI Commands
```bash
# Start STDIO server
praisonai mcp serve --transport stdio
# Start HTTP Stream server
praisonai mcp serve --transport http-stream --port 8080
# Start with authentication
praisonai mcp serve --transport http-stream --api-key YOUR_KEY
# List available tools
praisonai mcp list-tools
# List available resources
praisonai mcp list-resources
# List available prompts
praisonai mcp list-prompts
# Generate client configuration
praisonai mcp config-generate --client claude-desktop
# Check server health
praisonai mcp doctor
```
## Client Configuration
### Claude Desktop
Add to `~/.config/claude/claude_desktop_config.json`:
```json
{
"mcpServers": {
"praisonai": {
"command": "praisonai",
"args": ["mcp", "serve", "--transport", "stdio"]
}
}
}
```
### Cursor
Add to Cursor settings:
```json
{
"mcpServers": {
"praisonai": {
"command": "praisonai",
"args": ["mcp", "serve", "--transport", "stdio"]
}
}
}
```
## Environment Variables
Set the following environment variables for full functionality:
```bash
export OPENAI_API_KEY=your_key
export ANTHROPIC_API_KEY=your_key # Optional
export GOOGLE_API_KEY=your_key # Optional
```
## Available Tools
Run `praisonai mcp list-tools` to see all available tools, including:
- `praisonai.chat.completion` - Chat with LLM
- `praisonai.agent.chat` - Chat with an agent
- `praisonai.agent.run` - Run a task with an agent
- `praisonai.workflow.run` - Run a workflow
- `praisonai.images.generate` - Generate images
- `praisonai.audio.transcribe` - Transcribe audio
- `praisonai.embed.create` - Create embeddings
- `praisonai.memory.*` - Memory operations
- `praisonai.knowledge.*` - Knowledge base operations
- And many more...
## Available Resources
- `praisonai://memory/sessions` - List memory sessions
- `praisonai://workflows` - List available workflows
- `praisonai://tools` - List available tools
- `praisonai://agents` - List agent configurations
- `praisonai://knowledge/sources` - List knowledge sources
- `praisonai://config` - Get current configuration
- `praisonai://mcp/status` - Get MCP server status
## Available Prompts
- `deep-research` - Generate deep research prompts
- `code-review` - Generate code review prompts
- `workflow-auto` - Generate workflow auto-generation prompts
- `guardrail-check` - Generate guardrail check prompts
- `context-engineering` - Generate context engineering prompts
- `eval-criteria` - Generate evaluation criteria prompts
- `agent-instructions` - Generate agent instructions prompts
@@ -0,0 +1,71 @@
#!/usr/bin/env python3
"""
Custom Tools MCP Server Example
Register custom tools and expose them via MCP server.
Usage:
python custom_tools_server.py
"""
def main():
"""Run MCP server with custom tools."""
from praisonai.mcp_server.server import MCPServer
from praisonai.mcp_server.registry import register_tool, get_tool_registry
# Register custom tools
@register_tool("custom.greet")
def greet(name: str) -> str:
"""Greet a person by name."""
return f"Hello, {name}! Welcome to PraisonAI."
@register_tool("custom.calculate")
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression."""
try:
# Only allow safe math operations
allowed = set("0123456789+-*/.(). ")
if not all(c in allowed for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression) # Safe due to character filtering
return f"Result: {result}"
except Exception as e:
return f"Error: {e}"
@register_tool("custom.reverse")
def reverse_text(text: str) -> str:
"""Reverse a text string."""
return text[::-1]
@register_tool("custom.word_count")
def word_count(text: str) -> str:
"""Count words in text."""
words = text.split()
return f"Word count: {len(words)}"
# Show registered tools
registry = get_tool_registry()
tools = registry.list_all()
print(f"Registered {len(tools)} custom tools:")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
print()
# Create and run server
server = MCPServer(
name="praisonai-custom",
version="1.0.0",
instructions="Custom tools MCP server example.",
)
print("Starting MCP server on http://127.0.0.1:8080/mcp")
server.run(
transport="http-stream",
host="127.0.0.1",
port=8080,
)
if __name__ == "__main__":
main()
+44
View File
@@ -0,0 +1,44 @@
#!/usr/bin/env python3
"""
HTTP Stream MCP Server Example
Run PraisonAI as an MCP server using HTTP Stream transport.
This transport is useful for web-based integrations.
Usage:
python http_stream_server.py
Or via CLI:
praisonai mcp serve --transport http-stream --port 8080
"""
def main():
"""Run MCP server with HTTP Stream transport."""
from praisonai.mcp_server.server import MCPServer
from praisonai.mcp_server.adapters import register_all
# Register all tools, resources, and prompts
register_all()
# Create server
server = MCPServer(
name="praisonai",
version="1.0.0",
instructions="PraisonAI MCP Server - AI agent capabilities via MCP protocol.",
)
# Run with HTTP Stream transport
server.run(
transport="http-stream",
host="127.0.0.1",
port=8080,
endpoint="/mcp",
)
if __name__ == "__main__":
print("Starting PraisonAI MCP Server on http://127.0.0.1:8080/mcp")
print("Protocol Version: 2025-11-25")
print("Press Ctrl+C to stop")
main()
+238
View File
@@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""
MCP Client Example
Example of connecting to a PraisonAI MCP server as a client.
Prerequisites:
1. Start the MCP server first:
praisonai mcp serve --transport http-stream --port 8080
2. Then run this client:
python mcp_client_example.py
Environment Variables:
OPENAI_API_KEY - Required for chat completion tools
"""
import requests
def main():
"""Connect to MCP server and call tools."""
base_url = "http://127.0.0.1:8080/mcp"
session_id = None
print("PraisonAI MCP Client Example")
print("=" * 40)
# 1. Initialize connection
print("\n1. Initializing connection...")
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-11-25",
"capabilities": {},
"clientInfo": {
"name": "example-client",
"version": "1.0.0",
},
},
}
response = requests.post(
base_url,
json=init_request,
headers={"Content-Type": "application/json"},
)
if response.status_code == 200:
result = response.json()
session_id = response.headers.get("MCP-Session-Id")
print(f" Connected! Session ID: {session_id}")
print(f" Server: {result.get('result', {}).get('serverInfo', {})}")
print(f" Protocol: {result.get('result', {}).get('protocolVersion')}")
else:
print(f" Error: {response.status_code}")
return
# 2. Send initialized notification
print("\n2. Sending initialized notification...")
init_notification = {
"jsonrpc": "2.0",
"method": "notifications/initialized",
}
requests.post(
base_url,
json=init_notification,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
print(" Done")
# 3. List available tools
print("\n3. Listing available tools...")
list_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {},
}
response = requests.post(
base_url,
json=list_request,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
if response.status_code == 200:
result = response.json()
tools = result.get("result", {}).get("tools", [])
print(f" Found {len(tools)} tools")
for tool in tools[:5]: # Show first 5
print(f" - {tool.get('name')}")
if len(tools) > 5:
print(f" ... and {len(tools) - 5} more")
# 4. List available resources
print("\n4. Listing available resources...")
list_resources = {
"jsonrpc": "2.0",
"id": 3,
"method": "resources/list",
"params": {},
}
response = requests.post(
base_url,
json=list_resources,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
if response.status_code == 200:
result = response.json()
resources = result.get("result", {}).get("resources", [])
print(f" Found {len(resources)} resources")
for res in resources:
print(f" - {res.get('uri')}")
# 5. List available prompts
print("\n5. Listing available prompts...")
list_prompts = {
"jsonrpc": "2.0",
"id": 4,
"method": "prompts/list",
"params": {},
}
response = requests.post(
base_url,
json=list_prompts,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
if response.status_code == 200:
result = response.json()
prompts = result.get("result", {}).get("prompts", [])
print(f" Found {len(prompts)} prompts")
for prompt in prompts:
print(f" - {prompt.get('name')}")
# 6. Read a resource
print("\n6. Reading MCP status resource...")
read_resource = {
"jsonrpc": "2.0",
"id": 5,
"method": "resources/read",
"params": {
"uri": "praisonai://mcp/status",
},
}
response = requests.post(
base_url,
json=read_resource,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
if response.status_code == 200:
result = response.json()
contents = result.get("result", {}).get("contents", [])
if contents:
print(f" Status: {contents[0].get('text', 'N/A')}")
# 7. Get a prompt
print("\n7. Getting deep-research prompt...")
get_prompt = {
"jsonrpc": "2.0",
"id": 6,
"method": "prompts/get",
"params": {
"name": "deep-research",
"arguments": {
"topic": "AI agents",
"depth": "medium",
},
},
}
response = requests.post(
base_url,
json=get_prompt,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
if response.status_code == 200:
result = response.json()
messages = result.get("result", {}).get("messages", [])
if messages:
content = messages[0].get("content", {})
text = content.get("text", "") if isinstance(content, dict) else str(content)
print(f" Prompt preview: {text[:100]}...")
# 8. Ping
print("\n8. Pinging server...")
ping_request = {
"jsonrpc": "2.0",
"id": 7,
"method": "ping",
"params": {},
}
response = requests.post(
base_url,
json=ping_request,
headers={
"Content-Type": "application/json",
"MCP-Session-Id": session_id,
},
)
if response.status_code == 200:
print(" Pong!")
print("\n" + "=" * 40)
print("MCP client example completed successfully!")
if __name__ == "__main__":
main()
+36
View File
@@ -0,0 +1,36 @@
#!/usr/bin/env python3
"""
STDIO MCP Server Example
Run PraisonAI as an MCP server using STDIO transport.
This is the recommended transport for Claude Desktop integration.
Usage:
python stdio_server.py
Or via CLI:
praisonai mcp serve --transport stdio
"""
def main():
"""Run MCP server with STDIO transport."""
from praisonai.mcp_server.server import MCPServer
from praisonai.mcp_server.adapters import register_all
# Register all tools, resources, and prompts
register_all()
# Create and run server
server = MCPServer(
name="praisonai",
version="1.0.0",
instructions="PraisonAI MCP Server - AI agent capabilities via MCP protocol.",
)
# Run with STDIO transport
server.run(transport="stdio")
if __name__ == "__main__":
main()
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.7.6.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.7.6.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.7.7.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.7.7.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+45 -28
View File
@@ -350,22 +350,32 @@ Launch PraisonAI servers with unified discovery support.
return self.EXIT_SUCCESS
def cmd_mcp(self, args: List[str]) -> int:
"""Launch MCP server."""
spec = {
"host": {"default": self.DEFAULT_HOST},
"port": {"default": 8080, "type": "int"},
"transport": {"default": "http"},
"tools_file": {"default": None},
}
parsed = self._parse_args(args, spec)
"""Launch MCP server (DEPRECATED - use 'praisonai mcp serve' instead)."""
import sys
# Print deprecation warning
print("\n[yellow]⚠ DEPRECATION WARNING:[/yellow]", file=sys.stderr)
print("[yellow]'praisonai serve mcp' is deprecated and will be removed in a future version.[/yellow]", file=sys.stderr)
print("[yellow]Please use 'praisonai mcp serve' instead.[/yellow]\n", file=sys.stderr)
# Redirect to new MCP server CLI
try:
self._print_success(f"Starting MCP server on {parsed['host']}:{parsed['port']}")
print(f" Transport: {parsed['transport']}")
print(" Discovery: /__praisonai__/discovery")
from praisonai.mcp_server.cli import handle_mcp_command
app = self._create_mcp_app(parsed)
self._run_server(app, parsed["host"], parsed["port"], False)
# Convert args to new format
new_args = ["serve"]
spec = {
"host": {"default": self.DEFAULT_HOST},
"port": {"default": 8080, "type": "int"},
"transport": {"default": "http-stream"},
}
parsed = self._parse_args(args, spec)
new_args.extend(["--host", parsed["host"]])
new_args.extend(["--port", str(parsed["port"])])
new_args.extend(["--transport", parsed["transport"]])
return handle_mcp_command(new_args)
except ImportError as e:
self._print_error(f"Missing dependency: {e}")
@@ -373,8 +383,6 @@ Launch PraisonAI servers with unified discovery support.
except Exception as e:
self._print_error(str(e))
return self.EXIT_GENERAL_ERROR
return self.EXIT_SUCCESS
def _create_mcp_app(self, config: Dict[str, Any]) -> Any:
"""Create FastAPI app for MCP server."""
@@ -429,20 +437,31 @@ Launch PraisonAI servers with unified discovery support.
return app
def cmd_tools(self, args: List[str]) -> int:
"""Launch tools as MCP server."""
spec = {
"host": {"default": self.DEFAULT_HOST},
"port": {"default": 8081, "type": "int"},
"tools_file": {"default": None},
}
parsed = self._parse_args(args, spec)
"""Launch tools as MCP server (DEPRECATED - use 'praisonai mcp serve' instead)."""
import sys
# Print deprecation warning
print("\n[yellow]⚠ DEPRECATION WARNING:[/yellow]", file=sys.stderr)
print("[yellow]'praisonai serve tools' is deprecated and will be removed in a future version.[/yellow]", file=sys.stderr)
print("[yellow]Please use 'praisonai mcp serve' instead.[/yellow]\n", file=sys.stderr)
# Redirect to new MCP server CLI
try:
self._print_success(f"Starting tools MCP server on {parsed['host']}:{parsed['port']}")
print(" Discovery: /__praisonai__/discovery")
from praisonai.mcp_server.cli import handle_mcp_command
app = self._create_tools_app(parsed)
self._run_server(app, parsed["host"], parsed["port"], False)
# Convert args to new format
new_args = ["serve"]
spec = {
"host": {"default": self.DEFAULT_HOST},
"port": {"default": 8080, "type": "int"},
}
parsed = self._parse_args(args, spec)
new_args.extend(["--host", parsed["host"]])
new_args.extend(["--port", str(parsed["port"])])
new_args.extend(["--transport", "http-stream"])
return handle_mcp_command(new_args)
except ImportError as e:
self._print_error(f"Missing dependency: {e}")
@@ -450,8 +469,6 @@ Launch PraisonAI servers with unified discovery support.
except Exception as e:
self._print_error(str(e))
return self.EXIT_GENERAL_ERROR
return self.EXIT_SUCCESS
def _create_tools_app(self, config: Dict[str, Any]) -> Any:
"""Create FastAPI app for tools MCP server."""
+27 -1
View File
@@ -2938,8 +2938,32 @@ class PraisonAI:
else:
print(f"[red]MCP config not found: {config_name}[/red]")
elif action == 'serve':
# Start MCP server
from praisonai.mcp_server.cli import handle_mcp_command as mcp_serve_handler
exit_code = mcp_serve_handler(['serve'] + action_args)
sys.exit(exit_code)
elif action == 'list-tools':
# List MCP tools
from praisonai.mcp_server.cli import handle_mcp_command as mcp_serve_handler
exit_code = mcp_serve_handler(['list-tools'] + action_args)
sys.exit(exit_code)
elif action == 'config-generate':
# Generate client config
from praisonai.mcp_server.cli import handle_mcp_command as mcp_serve_handler
exit_code = mcp_serve_handler(['config-generate'] + action_args)
sys.exit(exit_code)
elif action == 'help' or action == '--help':
print("[bold]MCP Commands:[/bold]")
print("\n[bold cyan]Server Commands:[/bold cyan]")
print(" praisonai mcp serve - Start MCP server (STDIO)")
print(" praisonai mcp serve --transport http-stream - Start HTTP Stream server")
print(" praisonai mcp list-tools - List available MCP tools")
print(" praisonai mcp config-generate --client claude-desktop - Generate client config")
print("\n[bold cyan]Config Management:[/bold cyan]")
print(" praisonai mcp list - List all MCP configs")
print(" praisonai mcp show <name> - Show specific config")
print(" praisonai mcp create <name> <cmd> [args...] - Create a new config")
@@ -2948,7 +2972,9 @@ class PraisonAI:
print(" praisonai mcp disable <name> - Disable a config")
print("\n[bold]Config Location:[/bold]")
print(" .praison/mcp/*.json, ~/.praison/mcp/*.json")
print("\n[bold]Example:[/bold]")
print("\n[bold]Examples:[/bold]")
print(" praisonai mcp serve --transport stdio")
print(" praisonai mcp serve --transport http-stream --port 8080")
print(" praisonai mcp create filesystem npx -y @modelcontextprotocol/server-filesystem .")
else:
print(f"[red]Unknown mcp action: {action}[/red]")
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.7.6 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.7.7 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+8 -2
View File
@@ -39,12 +39,18 @@ class JobExecutor:
self.default_timeout = default_timeout
self.cleanup_interval = cleanup_interval
self._semaphore = asyncio.Semaphore(max_concurrent)
self._semaphore: Optional[asyncio.Semaphore] = None
self._running_tasks: Dict[str, asyncio.Task] = {}
self._cleanup_task: Optional[asyncio.Task] = None
self._shutdown = False
self._progress_callbacks: Dict[str, Callable] = {}
def _get_semaphore(self) -> asyncio.Semaphore:
"""Lazily create semaphore to avoid event loop issues."""
if self._semaphore is None:
self._semaphore = asyncio.Semaphore(self.max_concurrent)
return self._semaphore
async def start(self):
"""Start the executor and cleanup loop."""
self._shutdown = False
@@ -147,7 +153,7 @@ class JobExecutor:
async def _execute_job(self, job: Job):
"""Execute a job."""
async with self._semaphore:
async with self._get_semaphore():
try:
# Mark as running
job.start()
@@ -0,0 +1,58 @@
"""
PraisonAI MCP Server Module
Exposes PraisonAI capabilities as an MCP server that any MCP client can connect to.
Supports:
- STDIO transport (default, for Claude Desktop, Cursor, etc.)
- HTTP Stream transport (MCP 2025-03-26 spec)
Usage:
# STDIO mode (for Claude Desktop config)
praisonai mcp serve --transport stdio
# HTTP Stream mode
praisonai mcp serve --transport http-stream --port 8080
# Programmatic usage
from praisonai.mcp_server import MCPServer
server = MCPServer()
server.run(transport="stdio")
"""
__all__ = [
"MCPServer",
"MCPToolRegistry",
"MCPResourceRegistry",
"MCPPromptRegistry",
"register_tool",
"register_resource",
"register_prompt",
]
def __getattr__(name):
"""Lazy load server components."""
if name == "MCPServer":
from .server import MCPServer
return MCPServer
elif name == "MCPToolRegistry":
from .registry import MCPToolRegistry
return MCPToolRegistry
elif name == "MCPResourceRegistry":
from .registry import MCPResourceRegistry
return MCPResourceRegistry
elif name == "MCPPromptRegistry":
from .registry import MCPPromptRegistry
return MCPPromptRegistry
elif name == "register_tool":
from .registry import register_tool
return register_tool
elif name == "register_resource":
from .registry import register_resource
return register_resource
elif name == "register_prompt":
from .registry import register_prompt
return register_prompt
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,66 @@
"""
MCP Adapters
Adapters that convert PraisonAI capabilities, CLI features, and agent operations
into MCP tools, resources, and prompts.
"""
__all__ = [
"register_all",
"register_all_tools",
"register_capability_tools",
"register_extended_capability_tools",
"register_agent_tools",
"register_memory_tools",
"register_knowledge_tools",
"register_cli_tools",
"register_mcp_resources",
"register_mcp_prompts",
]
def __getattr__(name):
"""Lazy load adapters."""
if name == "register_all":
from .capabilities import register_all_tools
from .extended_capabilities import register_extended_capability_tools
from .cli_tools import register_cli_tools
from .resources import register_mcp_resources
from .prompts import register_mcp_prompts
def _register_all():
register_all_tools()
register_extended_capability_tools()
register_cli_tools()
register_mcp_resources()
register_mcp_prompts()
return _register_all
elif name == "register_all_tools":
from .capabilities import register_all_tools
return register_all_tools
elif name == "register_capability_tools":
from .capabilities import register_capability_tools
return register_capability_tools
elif name == "register_extended_capability_tools":
from .extended_capabilities import register_extended_capability_tools
return register_extended_capability_tools
elif name == "register_agent_tools":
from .agents import register_agent_tools
return register_agent_tools
elif name == "register_memory_tools":
from .memory import register_memory_tools
return register_memory_tools
elif name == "register_knowledge_tools":
from .knowledge import register_knowledge_tools
return register_knowledge_tools
elif name == "register_cli_tools":
from .cli_tools import register_cli_tools
return register_cli_tools
elif name == "register_mcp_resources":
from .resources import register_mcp_resources
return register_mcp_resources
elif name == "register_mcp_prompts":
from .prompts import register_mcp_prompts
return register_mcp_prompts
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,128 @@
"""
Agents Adapter
Maps PraisonAI agent operations to MCP tools.
"""
import logging
from typing import Optional
from ..registry import register_tool
logger = logging.getLogger(__name__)
def register_agent_tools() -> None:
"""Register agent-related MCP tools."""
@register_tool("praisonai.agent.chat")
def agent_chat(
message: str,
model: str = "gpt-4o-mini",
instructions: Optional[str] = None,
) -> str:
"""Chat with a PraisonAI agent."""
try:
from praisonaiagents import Agent
agent = Agent(
instructions=instructions or "You are a helpful assistant.",
llm=model,
)
result = agent.chat(message)
return str(result)
except ImportError:
return "Error: praisonaiagents not installed"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.agent.run")
def agent_run(
task: str,
model: str = "gpt-4o-mini",
instructions: Optional[str] = None,
) -> str:
"""Run a task with a PraisonAI agent."""
try:
from praisonaiagents import Agent
agent = Agent(
instructions=instructions or "You are a helpful assistant.",
llm=model,
)
result = agent.start(task)
return str(result)
except ImportError:
return "Error: praisonaiagents not installed"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.workflow.run")
def workflow_run(
yaml_content: str,
) -> str:
"""Run a workflow from YAML definition."""
try:
import yaml
from praisonai.agents_generator import AgentsGenerator
# Parse YAML to validate
yaml.safe_load(yaml_content)
# Write to temp file and run
import tempfile
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:
f.write(yaml_content)
temp_path = f.name
generator = AgentsGenerator(
agent_file=temp_path,
framework="praisonai",
)
result = generator.generate_crew_and_kickoff()
return str(result)
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.workflow.run_file")
def workflow_run_file(
file_path: str,
) -> str:
"""Run a workflow from YAML file path."""
try:
from praisonai.agents_generator import AgentsGenerator
generator = AgentsGenerator(
agent_file=file_path,
framework="praisonai",
)
result = generator.generate_crew_and_kickoff()
return str(result)
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.research.run")
def research_run(
topic: str,
depth: str = "medium",
) -> str:
"""Run deep research on a topic."""
try:
from praisonaiagents import Agent
instructions = f"""You are a research agent. Conduct thorough research on the given topic.
Research depth: {depth}
Provide comprehensive findings with sources where possible."""
agent = Agent(
instructions=instructions,
llm="gpt-4o",
)
result = agent.start(f"Research the following topic thoroughly: {topic}")
return str(result)
except ImportError:
return "Error: praisonaiagents not installed"
except Exception as e:
return f"Error: {e}"
logger.info("Registered agent MCP tools")
@@ -0,0 +1,168 @@
"""
Capabilities Adapter
Maps PraisonAI capabilities (audio, images, embeddings, etc.) to MCP tools.
"""
import logging
from typing import List, Optional
from ..registry import register_tool
logger = logging.getLogger(__name__)
def register_capability_tools() -> None:
"""Register all capability-based MCP tools."""
# Audio tools
@register_tool("praisonai.audio.transcribe")
def audio_transcribe(
file_path: str,
model: str = "whisper-1",
language: Optional[str] = None,
) -> str:
"""Transcribe audio file to text."""
from praisonai.capabilities import transcribe
result = transcribe(file=file_path, model=model, language=language)
return result.text if hasattr(result, "text") else str(result)
@register_tool("praisonai.audio.speech")
def audio_speech(
text: str,
model: str = "tts-1",
voice: str = "alloy",
) -> str:
"""Convert text to speech."""
from praisonai.capabilities import speech
result = speech(input=text, model=model, voice=voice)
return f"Audio generated: {result}"
# Image tools
@register_tool("praisonai.images.generate")
def images_generate(
prompt: str,
model: str = "dall-e-3",
size: str = "1024x1024",
n: int = 1,
) -> str:
"""Generate images from text prompt."""
from praisonai.capabilities import image_generate
result = image_generate(prompt=prompt, model=model, size=size, n=n)
if hasattr(result, "data"):
urls = [img.url for img in result.data if hasattr(img, "url")]
return f"Generated {len(urls)} image(s): {urls}"
return str(result)
# Embedding tools
@register_tool("praisonai.embed.create")
def embed_create(
text: str,
model: str = "text-embedding-3-small",
) -> str:
"""Create text embeddings."""
from praisonai.capabilities import embed
result = embed(input=text, model=model)
if hasattr(result, "data") and result.data:
return f"Embedding created with {len(result.data[0].embedding)} dimensions"
return str(result)
# Moderation tools
@register_tool("praisonai.moderate.check")
def moderate_check(text: str) -> str:
"""Check content for policy violations."""
from praisonai.capabilities import moderate
result = moderate(input=text)
if hasattr(result, "results") and result.results:
flagged = result.results[0].flagged
return f"Content flagged: {flagged}"
return str(result)
# Chat completion
@register_tool("praisonai.chat.completion")
def chat_completion(
message: str,
model: str = "gpt-4o-mini",
system_prompt: Optional[str] = None,
) -> str:
"""Generate chat completion."""
from praisonai.capabilities import chat_completion
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": message})
result = chat_completion(model=model, messages=messages)
if hasattr(result, "choices") and result.choices:
return result.choices[0].message.content
return str(result)
# Rerank
@register_tool("praisonai.rerank")
def rerank_documents(
query: str,
documents: List[str],
model: str = "rerank-english-v2.0",
top_n: int = 5,
) -> str:
"""Rerank documents by relevance to query."""
from praisonai.capabilities import rerank
result = rerank(query=query, documents=documents, model=model, top_n=top_n)
return str(result)
# RAG
@register_tool("praisonai.rag.query")
def rag_query(
query: str,
collection: str = "default",
) -> str:
"""Query RAG knowledge base."""
from praisonai.capabilities import rag_query
result = rag_query(query=query, collection=collection)
return str(result)
# Search
@register_tool("praisonai.search")
def search_web(query: str) -> str:
"""Search the web."""
from praisonai.capabilities import search
result = search(query=query)
return str(result)
# Guardrails
@register_tool("praisonai.guardrails.check")
def guardrails_check(
text: str,
guardrail_name: str = "default",
) -> str:
"""Apply guardrail to text."""
from praisonai.capabilities import apply_guardrail
result = apply_guardrail(text=text, guardrail_name=guardrail_name)
return str(result)
logger.info("Registered capability MCP tools")
def register_all_tools() -> None:
"""Register all MCP tools from all adapters."""
register_capability_tools()
# Import and register other adapters
try:
from .agents import register_agent_tools
register_agent_tools()
except ImportError:
logger.debug("Agent tools adapter not available")
try:
from .memory import register_memory_tools
register_memory_tools()
except ImportError:
logger.debug("Memory tools adapter not available")
try:
from .knowledge import register_knowledge_tools
register_knowledge_tools()
except ImportError:
logger.debug("Knowledge tools adapter not available")
logger.info("All MCP tools registered")
@@ -0,0 +1,568 @@
"""
CLI Tools Adapter
Maps PraisonAI CLI features to MCP tools:
- Workflow management
- Rules management
- Hooks management
- Session management
- Todo management
- Docs management
- Schedule management
- Profile management
- Deploy management
"""
import logging
from typing import Optional
from ..registry import register_tool
logger = logging.getLogger(__name__)
def register_cli_tools() -> None:
"""Register CLI-based MCP tools."""
# Workflow tools
@register_tool("praisonai.workflow.list")
def workflow_list() -> str:
"""List available workflows."""
try:
import glob
workflows = []
for pattern in ["*.yaml", "*.yml", "agents*.yaml", "workflow*.yaml"]:
workflows.extend(glob.glob(pattern))
if not workflows:
return "No workflows found in current directory"
return f"Available workflows: {workflows}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.workflow.validate")
def workflow_validate(file_path: str) -> str:
"""Validate a workflow YAML file."""
try:
import yaml
with open(file_path, 'r') as f:
config = yaml.safe_load(f)
required = ["framework", "topic"]
missing = [k for k in required if k not in config]
if missing:
return f"Invalid workflow: missing {missing}"
return f"Workflow valid: {file_path}"
except yaml.YAMLError as e:
return f"YAML error: {e}"
except FileNotFoundError:
return f"File not found: {file_path}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.workflow.show")
def workflow_show(file_path: str) -> str:
"""Show workflow configuration."""
try:
with open(file_path, 'r') as f:
content = f.read()
return content
except FileNotFoundError:
return f"File not found: {file_path}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.workflow.auto")
def workflow_auto(topic: str, pattern: str = "sequential") -> str:
"""Auto-generate a workflow for a topic."""
try:
from praisonai.auto import AutoGenerator
generator = AutoGenerator(topic=topic)
result = generator.generate(pattern=pattern)
return str(result)
except ImportError:
return "Error: AutoGenerator not available"
except Exception as e:
return f"Error: {e}"
# Rules tools
@register_tool("praisonai.rules.list")
def rules_list() -> str:
"""List active rules."""
try:
import os
rules_dir = os.path.expanduser("~/.praison/rules")
if not os.path.exists(rules_dir):
return "No rules directory found"
rules = os.listdir(rules_dir)
return f"Active rules: {rules}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.rules.show")
def rules_show(rule_name: str) -> str:
"""Show a specific rule."""
try:
import os
rule_path = os.path.expanduser(f"~/.praison/rules/{rule_name}")
if not os.path.exists(rule_path):
return f"Rule not found: {rule_name}"
with open(rule_path, 'r') as f:
content = f.read()
return content
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.rules.create")
def rules_create(rule_name: str, content: str) -> str:
"""Create a new rule."""
try:
import os
rules_dir = os.path.expanduser("~/.praison/rules")
os.makedirs(rules_dir, exist_ok=True)
rule_path = os.path.join(rules_dir, rule_name)
with open(rule_path, 'w') as f:
f.write(content)
return f"Rule created: {rule_name}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.rules.delete")
def rules_delete(rule_name: str) -> str:
"""Delete a rule."""
try:
import os
rule_path = os.path.expanduser(f"~/.praison/rules/{rule_name}")
if not os.path.exists(rule_path):
return f"Rule not found: {rule_name}"
os.remove(rule_path)
return f"Rule deleted: {rule_name}"
except Exception as e:
return f"Error: {e}"
# Hooks tools
@register_tool("praisonai.hooks.list")
def hooks_list() -> str:
"""List registered hooks."""
try:
from praisonaiagents.hooks import get_hook_manager
manager = get_hook_manager()
hooks = manager.list_hooks()
return str(hooks)
except ImportError:
return "Error: Hooks not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.hooks.stats")
def hooks_stats() -> str:
"""Get hook execution statistics."""
try:
from praisonaiagents.hooks import get_hook_manager
manager = get_hook_manager()
stats = manager.get_stats()
return str(stats)
except ImportError:
return "Error: Hooks not available"
except Exception as e:
return f"Error: {e}"
# Session tools
@register_tool("praisonai.session.list")
def session_list() -> str:
"""List active sessions."""
try:
from praisonaiagents.session import SessionManager
manager = SessionManager()
sessions = manager.list_sessions()
return str(sessions)
except ImportError:
return "Error: Session management not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.session.info")
def session_info(session_id: str) -> str:
"""Get session information."""
try:
from praisonaiagents.session import SessionManager
manager = SessionManager()
info = manager.get_session(session_id)
return str(info)
except ImportError:
return "Error: Session management not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.session.delete")
def session_delete(session_id: str) -> str:
"""Delete a session."""
try:
from praisonaiagents.session import SessionManager
manager = SessionManager()
manager.delete_session(session_id)
return f"Session deleted: {session_id}"
except ImportError:
return "Error: Session management not available"
except Exception as e:
return f"Error: {e}"
# Todo tools
@register_tool("praisonai.todo.list")
def todo_list() -> str:
"""List todo items."""
try:
import os
import json
todo_path = os.path.expanduser("~/.praison/todo.json")
if not os.path.exists(todo_path):
return "No todos found"
with open(todo_path, 'r') as f:
todos = json.load(f)
return str(todos)
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.todo.add")
def todo_add(content: str, priority: str = "medium") -> str:
"""Add a todo item."""
try:
import os
import json
import uuid
todo_path = os.path.expanduser("~/.praison/todo.json")
os.makedirs(os.path.dirname(todo_path), exist_ok=True)
todos = []
if os.path.exists(todo_path):
with open(todo_path, 'r') as f:
todos = json.load(f)
todo = {
"id": str(uuid.uuid4())[:8],
"content": content,
"priority": priority,
"status": "pending",
}
todos.append(todo)
with open(todo_path, 'w') as f:
json.dump(todos, f, indent=2)
return f"Todo added: {todo['id']}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.todo.complete")
def todo_complete(todo_id: str) -> str:
"""Mark a todo as complete."""
try:
import os
import json
todo_path = os.path.expanduser("~/.praison/todo.json")
if not os.path.exists(todo_path):
return "No todos found"
with open(todo_path, 'r') as f:
todos = json.load(f)
for todo in todos:
if todo.get("id") == todo_id:
todo["status"] = "completed"
break
else:
return f"Todo not found: {todo_id}"
with open(todo_path, 'w') as f:
json.dump(todos, f, indent=2)
return f"Todo completed: {todo_id}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.todo.delete")
def todo_delete(todo_id: str) -> str:
"""Delete a todo item."""
try:
import os
import json
todo_path = os.path.expanduser("~/.praison/todo.json")
if not os.path.exists(todo_path):
return "No todos found"
with open(todo_path, 'r') as f:
todos = json.load(f)
todos = [t for t in todos if t.get("id") != todo_id]
with open(todo_path, 'w') as f:
json.dump(todos, f, indent=2)
return f"Todo deleted: {todo_id}"
except Exception as e:
return f"Error: {e}"
# Tools discovery
@register_tool("praisonai.tools.list")
def tools_list() -> str:
"""List available tools."""
try:
from praisonaiagents.tools import get_available_tools
tools = get_available_tools()
return str(tools)
except ImportError:
return "Error: Tools module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.tools.info")
def tools_info(tool_name: str) -> str:
"""Get information about a tool."""
try:
from praisonaiagents.tools import get_tool_info
info = get_tool_info(tool_name)
return str(info)
except ImportError:
return "Error: Tools module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.tools.search")
def tools_search(query: str) -> str:
"""Search for tools by name or description."""
try:
from praisonaiagents.tools import search_tools
results = search_tools(query)
return str(results)
except ImportError:
return "Error: Tools module not available"
except Exception as e:
return f"Error: {e}"
# Schedule tools
@register_tool("praisonai.schedule.list")
def schedule_list() -> str:
"""List scheduled tasks."""
try:
from praisonai.agent_scheduler import AgentScheduler
scheduler = AgentScheduler()
tasks = scheduler.list_tasks()
return str(tasks)
except ImportError:
return "Error: Scheduler not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.schedule.add")
def schedule_add(
task_name: str,
cron: str,
workflow_path: str,
) -> str:
"""Add a scheduled task."""
try:
from praisonai.agent_scheduler import AgentScheduler
scheduler = AgentScheduler()
scheduler.add_task(task_name, cron, workflow_path)
return f"Task scheduled: {task_name}"
except ImportError:
return "Error: Scheduler not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.schedule.remove")
def schedule_remove(task_name: str) -> str:
"""Remove a scheduled task."""
try:
from praisonai.agent_scheduler import AgentScheduler
scheduler = AgentScheduler()
scheduler.remove_task(task_name)
return f"Task removed: {task_name}"
except ImportError:
return "Error: Scheduler not available"
except Exception as e:
return f"Error: {e}"
# Deploy tools
@register_tool("praisonai.deploy.validate")
def deploy_validate(config_path: str = "deploy.yaml") -> str:
"""Validate deployment configuration."""
try:
import yaml
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
required = ["name", "type"]
missing = [k for k in required if k not in config]
if missing:
return f"Invalid config: missing {missing}"
return f"Deployment config valid: {config_path}"
except FileNotFoundError:
return f"Config not found: {config_path}"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.deploy.status")
def deploy_status(deployment_name: Optional[str] = None) -> str:
"""Get deployment status."""
try:
from praisonai.deploy import get_deployment_status
status = get_deployment_status(deployment_name)
return str(status)
except ImportError:
return "Error: Deploy module not available"
except Exception as e:
return f"Error: {e}"
# Doctor/diagnostics tools
@register_tool("praisonai.doctor.env")
def doctor_env() -> str:
"""Check environment configuration."""
try:
import os
checks = {
"OPENAI_API_KEY": bool(os.environ.get("OPENAI_API_KEY")),
"ANTHROPIC_API_KEY": bool(os.environ.get("ANTHROPIC_API_KEY")),
"GOOGLE_API_KEY": bool(os.environ.get("GOOGLE_API_KEY")),
}
return str(checks)
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.doctor.config")
def doctor_config() -> str:
"""Check configuration files."""
try:
import os
config_files = [
"agents.yaml",
"config.yaml",
".env",
"pyproject.toml",
]
results = {}
for f in config_files:
results[f] = os.path.exists(f)
return str(results)
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.doctor.mcp")
def doctor_mcp() -> str:
"""Check MCP server configuration."""
try:
from ..registry import get_tool_registry, get_resource_registry, get_prompt_registry
tools = get_tool_registry().list_all()
resources = get_resource_registry().list_all()
prompts = get_prompt_registry().list_all()
return f"MCP Status: {len(tools)} tools, {len(resources)} resources, {len(prompts)} prompts"
except Exception as e:
return f"Error: {e}"
# Eval tools
@register_tool("praisonai.eval.accuracy")
def eval_accuracy(
agent_config: str,
input_text: str,
expected_output: str,
iterations: int = 3,
) -> str:
"""Run accuracy evaluation on an agent."""
try:
from praisonaiagents.eval import AccuracyEval
from praisonaiagents import Agent
agent = Agent.from_yaml(agent_config)
eval_runner = AccuracyEval(
agent=agent,
input=input_text,
expected_output=expected_output,
num_iterations=iterations,
)
result = eval_runner.run()
return str(result)
except ImportError:
return "Error: Eval module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.eval.performance")
def eval_performance(
agent_config: str,
input_text: str,
iterations: int = 10,
) -> str:
"""Run performance evaluation on an agent."""
try:
from praisonaiagents.eval import PerformanceEval
from praisonaiagents import Agent
agent = Agent.from_yaml(agent_config)
eval_runner = PerformanceEval(
func=lambda: agent.chat(input_text),
num_iterations=iterations,
)
result = eval_runner.run()
return str(result)
except ImportError:
return "Error: Eval module not available"
except Exception as e:
return f"Error: {e}"
# MCP config management
@register_tool("praisonai.mcp_config.list")
def mcp_config_list() -> str:
"""List MCP server configurations."""
try:
from praisonaiagents.memory.mcp_config import MCPConfigManager
manager = MCPConfigManager()
configs = manager.list_configs()
return str(configs)
except ImportError:
return "Error: MCP config manager not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.mcp_config.show")
def mcp_config_show(config_name: str) -> str:
"""Show MCP server configuration."""
try:
from praisonaiagents.memory.mcp_config import MCPConfigManager
manager = MCPConfigManager()
config = manager.get_config(config_name)
return str(config)
except ImportError:
return "Error: MCP config manager not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.mcp_config.enable")
def mcp_config_enable(config_name: str) -> str:
"""Enable an MCP server configuration."""
try:
from praisonaiagents.memory.mcp_config import MCPConfigManager
manager = MCPConfigManager()
manager.enable_config(config_name)
return f"MCP config enabled: {config_name}"
except ImportError:
return "Error: MCP config manager not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.mcp_config.disable")
def mcp_config_disable(config_name: str) -> str:
"""Disable an MCP server configuration."""
try:
from praisonaiagents.memory.mcp_config import MCPConfigManager
manager = MCPConfigManager()
manager.disable_config(config_name)
return f"MCP config disabled: {config_name}"
except ImportError:
return "Error: MCP config manager not available"
except Exception as e:
return f"Error: {e}"
logger.info("Registered CLI MCP tools")
@@ -0,0 +1,462 @@
"""
Extended Capabilities Adapter
Maps additional PraisonAI capabilities to MCP tools:
- Videos
- Files
- Batches
- Vector Stores
- OCR
- Assistants
- Fine-tuning
- A2A
- Containers
- Skills
- Realtime
"""
import logging
from typing import List, Optional
from ..registry import register_tool
logger = logging.getLogger(__name__)
def register_extended_capability_tools() -> None:
"""Register extended capability-based MCP tools."""
# Video tools
@register_tool("praisonai.videos.generate")
def videos_generate(
prompt: str,
model: str = "runway-gen3",
duration: int = 4,
) -> str:
"""Generate video from text prompt."""
try:
from praisonai.capabilities import video_generate
result = video_generate(prompt=prompt, model=model, duration=duration)
return str(result)
except ImportError:
return "Error: Video generation not available"
except Exception as e:
return f"Error: {e}"
# File tools
@register_tool("praisonai.files.create")
def files_create(
file_path: str,
purpose: str = "assistants",
) -> str:
"""Upload a file."""
try:
from praisonai.capabilities import file_create
result = file_create(file=file_path, purpose=purpose)
if hasattr(result, "id"):
return f"File created: {result.id}"
return str(result)
except ImportError:
return "Error: File operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.files.list")
def files_list(purpose: Optional[str] = None) -> str:
"""List uploaded files."""
try:
from praisonai.capabilities import file_list
result = file_list(purpose=purpose)
if hasattr(result, "data"):
files = [{"id": f.id, "filename": f.filename} for f in result.data]
return str(files)
return str(result)
except ImportError:
return "Error: File operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.files.retrieve")
def files_retrieve(file_id: str) -> str:
"""Retrieve file metadata."""
try:
from praisonai.capabilities import file_retrieve
result = file_retrieve(file_id=file_id)
return str(result)
except ImportError:
return "Error: File operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.files.delete")
def files_delete(file_id: str) -> str:
"""Delete a file."""
try:
from praisonai.capabilities import file_delete
file_delete(file_id=file_id)
return f"File deleted: {file_id}"
except ImportError:
return "Error: File operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.files.content")
def files_content(file_id: str) -> str:
"""Get file content."""
try:
from praisonai.capabilities import file_content
result = file_content(file_id=file_id)
return str(result)
except ImportError:
return "Error: File operations not available"
except Exception as e:
return f"Error: {e}"
# Batch tools
@register_tool("praisonai.batches.create")
def batches_create(
input_file_id: str,
endpoint: str = "/v1/chat/completions",
completion_window: str = "24h",
) -> str:
"""Create a batch processing job."""
try:
from praisonai.capabilities import batch_create
result = batch_create(
input_file_id=input_file_id,
endpoint=endpoint,
completion_window=completion_window,
)
if hasattr(result, "id"):
return f"Batch created: {result.id}"
return str(result)
except ImportError:
return "Error: Batch operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.batches.list")
def batches_list() -> str:
"""List batch jobs."""
try:
from praisonai.capabilities import batch_list
result = batch_list()
return str(result)
except ImportError:
return "Error: Batch operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.batches.retrieve")
def batches_retrieve(batch_id: str) -> str:
"""Retrieve batch job status."""
try:
from praisonai.capabilities import batch_retrieve
result = batch_retrieve(batch_id=batch_id)
return str(result)
except ImportError:
return "Error: Batch operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.batches.cancel")
def batches_cancel(batch_id: str) -> str:
"""Cancel a batch job."""
try:
from praisonai.capabilities import batch_cancel
batch_cancel(batch_id=batch_id)
return f"Batch cancelled: {batch_id}"
except ImportError:
return "Error: Batch operations not available"
except Exception as e:
return f"Error: {e}"
# Vector Store tools
@register_tool("praisonai.vector_stores.create")
def vector_stores_create(
name: str,
file_ids: Optional[List[str]] = None,
) -> str:
"""Create a vector store."""
try:
from praisonai.capabilities import vector_store_create
result = vector_store_create(name=name, file_ids=file_ids or [])
if hasattr(result, "id"):
return f"Vector store created: {result.id}"
return str(result)
except ImportError:
return "Error: Vector store operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.vector_stores.search")
def vector_stores_search(
vector_store_id: str,
query: str,
max_results: int = 10,
) -> str:
"""Search a vector store."""
try:
from praisonai.capabilities import vector_store_search
result = vector_store_search(
vector_store_id=vector_store_id,
query=query,
max_num_results=max_results,
)
return str(result)
except ImportError:
return "Error: Vector store operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.vector_stores.file_create")
def vector_stores_file_create(
vector_store_id: str,
file_id: str,
) -> str:
"""Add a file to a vector store."""
try:
from praisonai.capabilities import vector_store_file_create
vector_store_file_create(
vector_store_id=vector_store_id,
file_id=file_id,
)
return f"File {file_id} added to vector store {vector_store_id}"
except ImportError:
return "Error: Vector store operations not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.vector_stores.file_list")
def vector_stores_file_list(vector_store_id: str) -> str:
"""List files in a vector store."""
try:
from praisonai.capabilities import vector_store_file_list
result = vector_store_file_list(vector_store_id=vector_store_id)
return str(result)
except ImportError:
return "Error: Vector store operations not available"
except Exception as e:
return f"Error: {e}"
# OCR tools
@register_tool("praisonai.ocr")
def ocr_extract(
image_path: str,
model: str = "gpt-4o",
) -> str:
"""Extract text from image using OCR."""
try:
from praisonai.capabilities import ocr
result = ocr(file=image_path, model=model)
return str(result)
except ImportError:
return "Error: OCR not available"
except Exception as e:
return f"Error: {e}"
# Assistant tools
@register_tool("praisonai.assistants.create")
def assistants_create(
name: str,
instructions: str,
model: str = "gpt-4o",
tools: Optional[List[str]] = None,
) -> str:
"""Create an OpenAI-style assistant."""
try:
from praisonai.capabilities import assistant_create
result = assistant_create(
name=name,
instructions=instructions,
model=model,
tools=tools or [],
)
if hasattr(result, "id"):
return f"Assistant created: {result.id}"
return str(result)
except ImportError:
return "Error: Assistants not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.assistants.list")
def assistants_list() -> str:
"""List assistants."""
try:
from praisonai.capabilities import assistant_list
result = assistant_list()
return str(result)
except ImportError:
return "Error: Assistants not available"
except Exception as e:
return f"Error: {e}"
# Fine-tuning tools
@register_tool("praisonai.fine_tuning.create")
def fine_tuning_create(
training_file: str,
model: str = "gpt-4o-mini",
suffix: Optional[str] = None,
) -> str:
"""Create a fine-tuning job."""
try:
from praisonai.capabilities import fine_tuning_create
result = fine_tuning_create(
training_file=training_file,
model=model,
suffix=suffix,
)
if hasattr(result, "id"):
return f"Fine-tuning job created: {result.id}"
return str(result)
except ImportError:
return "Error: Fine-tuning not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.fine_tuning.list")
def fine_tuning_list() -> str:
"""List fine-tuning jobs."""
try:
from praisonai.capabilities import fine_tuning_list
result = fine_tuning_list()
return str(result)
except ImportError:
return "Error: Fine-tuning not available"
except Exception as e:
return f"Error: {e}"
# A2A tools
@register_tool("praisonai.a2a.send")
def a2a_send(
agent_url: str,
message: str,
task_id: Optional[str] = None,
) -> str:
"""Send message to another agent via A2A protocol."""
try:
from praisonai.capabilities import a2a_send
result = a2a_send(
agent_url=agent_url,
message=message,
task_id=task_id,
)
return str(result)
except ImportError:
return "Error: A2A not available"
except Exception as e:
return f"Error: {e}"
# Container tools
@register_tool("praisonai.containers.create")
def containers_create(
name: str,
image: str = "python:3.11",
) -> str:
"""Create a container for code execution."""
try:
from praisonai.capabilities import container_create
result = container_create(name=name, image=image)
if hasattr(result, "id"):
return f"Container created: {result.id}"
return str(result)
except ImportError:
return "Error: Containers not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.containers.file_read")
def containers_file_read(
container_id: str,
file_path: str,
) -> str:
"""Read a file from a container."""
try:
from praisonai.capabilities import container_file_read
result = container_file_read(
container_id=container_id,
file_path=file_path,
)
return str(result)
except ImportError:
return "Error: Containers not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.containers.file_write")
def containers_file_write(
container_id: str,
file_path: str,
content: str,
) -> str:
"""Write a file to a container."""
try:
from praisonai.capabilities import container_file_write
container_file_write(
container_id=container_id,
file_path=file_path,
content=content,
)
return f"File {file_path} written to container {container_id}"
except ImportError:
return "Error: Containers not available"
except Exception as e:
return f"Error: {e}"
# Skills tools
@register_tool("praisonai.skills.list")
def skills_list() -> str:
"""List available skills."""
try:
from praisonai.capabilities import skill_list
result = skill_list()
return str(result)
except ImportError:
return "Error: Skills not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.skills.load")
def skills_load(skill_path: str) -> str:
"""Load a skill from path."""
try:
from praisonai.capabilities import skill_load
result = skill_load(skill_path=skill_path)
return str(result)
except ImportError:
return "Error: Skills not available"
except Exception as e:
return f"Error: {e}"
# Realtime tools
@register_tool("praisonai.realtime.connect")
def realtime_connect(
model: str = "gpt-4o-realtime-preview",
) -> str:
"""Connect to realtime API."""
try:
from praisonai.capabilities import realtime_connect
result = realtime_connect(model=model)
return str(result)
except ImportError:
return "Error: Realtime not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.realtime.send")
def realtime_send(
session_id: str,
message: str,
) -> str:
"""Send message to realtime session."""
try:
from praisonai.capabilities import realtime_send
result = realtime_send(session_id=session_id, message=message)
return str(result)
except ImportError:
return "Error: Realtime not available"
except Exception as e:
return f"Error: {e}"
logger.info("Registered extended capability MCP tools")
@@ -0,0 +1,93 @@
"""
Knowledge Adapter
Maps PraisonAI knowledge operations to MCP tools.
"""
import logging
from ..registry import register_tool
logger = logging.getLogger(__name__)
def register_knowledge_tools() -> None:
"""Register knowledge-related MCP tools."""
@register_tool("praisonai.knowledge.add")
def knowledge_add(
source: str,
source_type: str = "text",
) -> str:
"""Add a knowledge source (text, file, or URL)."""
try:
from praisonaiagents.knowledge import Knowledge
knowledge = Knowledge()
knowledge.add(source, source_type=source_type)
return f"Knowledge added from {source_type}"
except ImportError:
return "Error: Knowledge module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.knowledge.query")
def knowledge_query(
query: str,
limit: int = 5,
) -> str:
"""Query the knowledge base."""
try:
from praisonaiagents.knowledge import Knowledge
knowledge = Knowledge()
results = knowledge.query(query, limit=limit)
return str(results)
except ImportError:
return "Error: Knowledge module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.knowledge.list")
def knowledge_list() -> str:
"""List all knowledge sources."""
try:
from praisonaiagents.knowledge import Knowledge
knowledge = Knowledge()
sources = knowledge.list_sources()
return str(sources)
except ImportError:
return "Error: Knowledge module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.knowledge.clear")
def knowledge_clear() -> str:
"""Clear all knowledge."""
try:
from praisonaiagents.knowledge import Knowledge
knowledge = Knowledge()
knowledge.clear()
return "Knowledge cleared"
except ImportError:
return "Error: Knowledge module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.knowledge.stats")
def knowledge_stats() -> str:
"""Get knowledge base statistics."""
try:
from praisonaiagents.knowledge import Knowledge
knowledge = Knowledge()
stats = knowledge.stats()
return str(stats)
except ImportError:
return "Error: Knowledge module not available"
except Exception as e:
return f"Error: {e}"
logger.info("Registered knowledge MCP tools")
@@ -0,0 +1,104 @@
"""
Memory Adapter
Maps PraisonAI memory operations to MCP tools.
"""
import logging
from typing import Optional
from ..registry import register_tool
logger = logging.getLogger(__name__)
def register_memory_tools() -> None:
"""Register memory-related MCP tools."""
@register_tool("praisonai.memory.show")
def memory_show(session_id: Optional[str] = None) -> str:
"""Show memory contents for a session."""
try:
from praisonaiagents.memory import Memory
memory = Memory()
if session_id:
data = memory.get_session(session_id)
else:
data = memory.get_all()
return str(data)
except ImportError:
return "Error: Memory module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.memory.add")
def memory_add(
content: str,
session_id: Optional[str] = None,
metadata: Optional[str] = None,
) -> str:
"""Add content to memory."""
try:
from praisonaiagents.memory import Memory
import json
memory = Memory()
meta = json.loads(metadata) if metadata else {}
memory.add(content, session_id=session_id, metadata=meta)
return "Memory added successfully"
except ImportError:
return "Error: Memory module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.memory.search")
def memory_search(
query: str,
limit: int = 10,
) -> str:
"""Search memory for relevant content."""
try:
from praisonaiagents.memory import Memory
memory = Memory()
results = memory.search(query, limit=limit)
return str(results)
except ImportError:
return "Error: Memory module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.memory.clear")
def memory_clear(session_id: Optional[str] = None) -> str:
"""Clear memory contents."""
try:
from praisonaiagents.memory import Memory
memory = Memory()
if session_id:
memory.clear_session(session_id)
return f"Session {session_id} cleared"
else:
memory.clear_all()
return "All memory cleared"
except ImportError:
return "Error: Memory module not available"
except Exception as e:
return f"Error: {e}"
@register_tool("praisonai.memory.sessions")
def memory_sessions() -> str:
"""List all memory sessions."""
try:
from praisonaiagents.memory import Memory
memory = Memory()
sessions = memory.list_sessions()
return str(sessions)
except ImportError:
return "Error: Memory module not available"
except Exception as e:
return f"Error: {e}"
logger.info("Registered memory MCP tools")
@@ -0,0 +1,306 @@
"""
MCP Prompts Adapter
Registers MCP prompts for PraisonAI:
- deep-research
- code-review
- workflow-auto
- guardrail-check
- context-engineering
- eval-criteria
"""
import logging
from ..registry import register_prompt
logger = logging.getLogger(__name__)
def register_mcp_prompts() -> None:
"""Register MCP prompts."""
@register_prompt(
"deep-research",
description="Generate a deep research prompt for comprehensive topic analysis",
arguments=[
{"name": "topic", "description": "Research topic", "required": True},
{"name": "depth", "description": "Research depth (shallow/medium/deep)", "required": False},
{"name": "focus_areas", "description": "Specific areas to focus on", "required": False},
],
)
def deep_research_prompt(
topic: str,
depth: str = "medium",
focus_areas: str = "",
) -> list:
"""Generate deep research prompt."""
depth_instructions = {
"shallow": "Provide a brief overview with key points.",
"medium": "Conduct thorough research with multiple perspectives and sources.",
"deep": "Perform exhaustive research covering all aspects, historical context, current state, and future implications.",
}
focus_text = f"\n\nFocus particularly on: {focus_areas}" if focus_areas else ""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Research the following topic comprehensively:
Topic: {topic}
Research Depth: {depth}
{depth_instructions.get(depth, depth_instructions["medium"])}
{focus_text}
Please provide:
1. Executive Summary
2. Key Findings
3. Detailed Analysis
4. Sources and References
5. Conclusions and Recommendations"""
}
}
]
@register_prompt(
"code-review",
description="Generate a code review prompt for analyzing code quality",
arguments=[
{"name": "code", "description": "Code to review", "required": True},
{"name": "language", "description": "Programming language", "required": False},
{"name": "focus", "description": "Review focus (security/performance/style/all)", "required": False},
],
)
def code_review_prompt(
code: str,
language: str = "auto",
focus: str = "all",
) -> list:
"""Generate code review prompt."""
focus_areas = {
"security": "Focus on security vulnerabilities, input validation, and potential exploits.",
"performance": "Focus on performance optimizations, algorithmic efficiency, and resource usage.",
"style": "Focus on code style, readability, naming conventions, and best practices.",
"all": "Review all aspects: security, performance, style, and correctness.",
}
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Please review the following {language} code:
```{language}
{code}
```
Review Focus: {focus}
{focus_areas.get(focus, focus_areas["all"])}
Please provide:
1. Overall Assessment
2. Issues Found (with severity: critical/major/minor)
3. Suggestions for Improvement
4. Positive Aspects
5. Refactored Code (if applicable)"""
}
}
]
@register_prompt(
"workflow-auto",
description="Generate a workflow auto-generation prompt",
arguments=[
{"name": "task", "description": "Task description", "required": True},
{"name": "pattern", "description": "Workflow pattern (sequential/parallel/routing)", "required": False},
{"name": "agents_count", "description": "Number of agents to use", "required": False},
],
)
def workflow_auto_prompt(
task: str,
pattern: str = "sequential",
agents_count: str = "auto",
) -> list:
"""Generate workflow auto-generation prompt."""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Generate a PraisonAI workflow configuration for the following task:
Task: {task}
Workflow Pattern: {pattern}
Number of Agents: {agents_count}
Please generate a YAML configuration that includes:
1. Framework specification
2. Topic/goal definition
3. Agent roles with:
- Role name
- Goal
- Backstory
- Tasks with descriptions and expected outputs
4. Appropriate tools for each agent
Output the complete agents.yaml configuration."""
}
}
]
@register_prompt(
"guardrail-check",
description="Generate a guardrail check prompt for content safety",
arguments=[
{"name": "content", "description": "Content to check", "required": True},
{"name": "rules", "description": "Specific rules to enforce", "required": False},
],
)
def guardrail_check_prompt(
content: str,
rules: str = "",
) -> list:
"""Generate guardrail check prompt."""
rules_text = f"\n\nAdditional Rules:\n{rules}" if rules else ""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Analyze the following content for safety and policy compliance:
Content:
{content}
{rules_text}
Please check for:
1. Harmful or dangerous content
2. Personal information exposure
3. Inappropriate language
4. Policy violations
5. Factual accuracy concerns
Provide:
- Safety Score (1-10)
- Issues Found
- Recommendations
- Modified Safe Version (if needed)"""
}
}
]
@register_prompt(
"context-engineering",
description="Generate a context engineering prompt for optimal LLM interaction",
arguments=[
{"name": "goal", "description": "What you want to achieve", "required": True},
{"name": "constraints", "description": "Any constraints or requirements", "required": False},
],
)
def context_engineering_prompt(
goal: str,
constraints: str = "",
) -> list:
"""Generate context engineering prompt."""
constraints_text = f"\n\nConstraints:\n{constraints}" if constraints else ""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Help me engineer an optimal prompt/context for the following goal:
Goal: {goal}
{constraints_text}
Please provide:
1. Optimal System Prompt
2. User Prompt Template
3. Few-shot Examples (if helpful)
4. Recommended Model Settings (temperature, etc.)
5. Expected Output Format
6. Potential Edge Cases to Handle"""
}
}
]
@register_prompt(
"eval-criteria",
description="Generate evaluation criteria for agent output assessment",
arguments=[
{"name": "task_type", "description": "Type of task being evaluated", "required": True},
{"name": "output_format", "description": "Expected output format", "required": False},
],
)
def eval_criteria_prompt(
task_type: str,
output_format: str = "text",
) -> list:
"""Generate evaluation criteria prompt."""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Generate comprehensive evaluation criteria for the following:
Task Type: {task_type}
Expected Output Format: {output_format}
Please provide:
1. Accuracy Criteria (what makes output correct)
2. Quality Metrics (completeness, clarity, relevance)
3. Format Compliance Checks
4. Edge Case Handling Assessment
5. Scoring Rubric (1-10 scale with descriptions)
6. Example Good/Bad Outputs"""
}
}
]
@register_prompt(
"agent-instructions",
description="Generate optimal agent instructions for a specific role",
arguments=[
{"name": "role", "description": "Agent role/persona", "required": True},
{"name": "capabilities", "description": "Available capabilities/tools", "required": False},
],
)
def agent_instructions_prompt(
role: str,
capabilities: str = "",
) -> list:
"""Generate agent instructions prompt."""
caps_text = f"\n\nAvailable Capabilities:\n{capabilities}" if capabilities else ""
return [
{
"role": "user",
"content": {
"type": "text",
"text": f"""Generate optimal instructions for an AI agent with the following role:
Role: {role}
{caps_text}
Please provide:
1. Core Identity Statement
2. Primary Objectives
3. Behavioral Guidelines
4. Communication Style
5. Decision-Making Framework
6. Error Handling Approach
7. Collaboration Guidelines (if working with other agents)"""
}
}
]
logger.info("Registered MCP prompts")
@@ -0,0 +1,124 @@
"""
MCP Resources Adapter
Registers MCP resources for PraisonAI:
- praisonai://memory/sessions
- praisonai://memory/{session_id}
- praisonai://workflows
- praisonai://tools
- praisonai://agents
- praisonai://knowledge/sources
"""
import logging
from ..registry import register_resource
logger = logging.getLogger(__name__)
def register_mcp_resources() -> None:
"""Register MCP resources."""
@register_resource("praisonai://memory/sessions")
def memory_sessions_resource() -> dict:
"""List all memory sessions."""
try:
from praisonaiagents.memory import Memory
memory = Memory()
sessions = memory.list_sessions()
return {"sessions": sessions}
except ImportError:
return {"error": "Memory module not available", "sessions": []}
except Exception as e:
return {"error": str(e), "sessions": []}
@register_resource("praisonai://workflows")
def workflows_resource() -> dict:
"""List available workflows in current directory."""
try:
import os
import glob
workflows = []
for pattern in ["*.yaml", "*.yml"]:
for f in glob.glob(pattern):
if "agent" in f.lower() or "workflow" in f.lower():
workflows.append(f)
return {"workflows": workflows, "cwd": os.getcwd()}
except Exception as e:
return {"error": str(e), "workflows": []}
@register_resource("praisonai://tools")
def tools_resource() -> dict:
"""List available tools."""
try:
from ..registry import get_tool_registry
registry = get_tool_registry()
tools = registry.list_schemas()
return {"tools": [t["name"] for t in tools], "count": len(tools)}
except Exception as e:
return {"error": str(e), "tools": []}
@register_resource("praisonai://agents")
def agents_resource() -> dict:
"""List agent configurations."""
try:
import os
import yaml
agents = []
for f in ["agents.yaml", "agents.yml"]:
if os.path.exists(f):
with open(f, 'r') as file:
config = yaml.safe_load(file)
if config and "roles" in config:
agents.extend(list(config["roles"].keys()))
return {"agents": agents}
except Exception as e:
return {"error": str(e), "agents": []}
@register_resource("praisonai://knowledge/sources")
def knowledge_sources_resource() -> dict:
"""List knowledge sources."""
try:
from praisonaiagents.knowledge import Knowledge
knowledge = Knowledge()
sources = knowledge.list_sources()
return {"sources": sources}
except ImportError:
return {"error": "Knowledge module not available", "sources": []}
except Exception as e:
return {"error": str(e), "sources": []}
@register_resource("praisonai://config")
def config_resource() -> dict:
"""Get current configuration."""
try:
import os
config = {
"openai_api_key_set": bool(os.environ.get("OPENAI_API_KEY")),
"anthropic_api_key_set": bool(os.environ.get("ANTHROPIC_API_KEY")),
"google_api_key_set": bool(os.environ.get("GOOGLE_API_KEY")),
"cwd": os.getcwd(),
}
return config
except Exception as e:
return {"error": str(e)}
@register_resource("praisonai://mcp/status")
def mcp_status_resource() -> dict:
"""Get MCP server status."""
try:
from ..registry import get_tool_registry, get_resource_registry, get_prompt_registry
from ..server import PROTOCOL_VERSION
return {
"protocol_version": PROTOCOL_VERSION,
"tools_count": len(get_tool_registry().list_all()),
"resources_count": len(get_resource_registry().list_all()),
"prompts_count": len(get_prompt_registry().list_all()),
"status": "healthy",
}
except Exception as e:
return {"error": str(e), "status": "error"}
logger.info("Registered MCP resources")
+517
View File
@@ -0,0 +1,517 @@
"""
MCP Server CLI Integration
Provides CLI commands for managing the MCP server:
- praisonai mcp serve --transport stdio|http-stream
- praisonai mcp list-tools
- praisonai mcp config-generate
"""
import argparse
import json
import logging
import sys
from typing import List
logger = logging.getLogger(__name__)
class MCPServerCLI:
"""CLI handler for MCP server commands."""
EXIT_SUCCESS = 0
EXIT_ERROR = 1
def __init__(self):
"""Initialize CLI handler."""
self._server = None
def handle(self, args: List[str]) -> int:
"""
Handle MCP CLI subcommand.
Args:
args: Command arguments
Returns:
Exit code
"""
if not args:
self._print_help()
return self.EXIT_SUCCESS
command = args[0]
remaining = args[1:]
commands = {
"serve": self.cmd_serve,
"list-tools": self.cmd_list_tools,
"list-resources": self.cmd_list_resources,
"list-prompts": self.cmd_list_prompts,
"config-generate": self.cmd_config_generate,
"doctor": self.cmd_doctor,
"help": lambda _: self._print_help() or self.EXIT_SUCCESS,
"--help": lambda _: self._print_help() or self.EXIT_SUCCESS,
"-h": lambda _: self._print_help() or self.EXIT_SUCCESS,
}
if command in commands:
return commands[command](remaining)
else:
self._print_error(f"Unknown command: {command}")
self._print_help()
return self.EXIT_ERROR
def _print_help(self) -> None:
"""Print help message."""
help_text = """
[bold cyan]PraisonAI MCP Server (Protocol Version 2025-11-25)[/bold cyan]
Run PraisonAI as an MCP server for Claude Desktop, Cursor, Windsurf, and other MCP clients.
[bold]Usage:[/bold]
praisonai mcp <command> [options]
[bold]Commands:[/bold]
serve Start the MCP server
list-tools List available MCP tools
list-resources List available MCP resources
list-prompts List available MCP prompts
config-generate Generate client configuration
doctor Check MCP server health
[bold]Serve Options:[/bold]
--transport <type> Transport: stdio (default) or http-stream
--host <host> HTTP host (default: 127.0.0.1)
--port <port> HTTP port (default: 8080)
--endpoint <path> HTTP endpoint (default: /mcp)
--api-key <key> API key for authentication
--name <name> Server name (default: praisonai)
--response-mode <mode> Response mode: batch (default) or stream
--cors-origins <list> Comma-separated CORS origins
--allowed-origins <list> Comma-separated allowed origins for security
--session-ttl <secs> Session TTL in seconds (default: 3600)
--no-termination Disable client session termination
--resumability Enable SSE resumability (default: true)
--log-level <level> Log level: debug, info, warning, error
--json Output in JSON format
[bold]Config Generate Options:[/bold]
--client <name> Client type: claude-desktop, cursor, vscode, windsurf
--output <path> Output file path
--transport <type> Transport for config
[bold]Examples:[/bold]
# Start STDIO server (for Claude Desktop)
praisonai mcp serve --transport stdio
# Start HTTP Stream server with authentication
praisonai mcp serve --transport http-stream --port 8080 --api-key mykey
# Start with custom allowed origins
praisonai mcp serve --transport http-stream --allowed-origins "http://localhost:3000"
# Generate Claude Desktop config
praisonai mcp config-generate --client claude-desktop
# List available tools
praisonai mcp list-tools
# Check MCP server health
praisonai mcp doctor
"""
self._print_rich(help_text)
def _print_rich(self, text: str) -> None:
"""Print with rich formatting if available."""
try:
from rich import print as rprint
rprint(text)
except ImportError:
import re
plain = re.sub(r'\[/?[^\]]+\]', '', text)
print(plain)
def _print_error(self, message: str) -> None:
"""Print error message."""
try:
from rich import print as rprint
rprint(f"[red]Error: {message}[/red]")
except ImportError:
print(f"Error: {message}", file=sys.stderr)
def _print_success(self, message: str) -> None:
"""Print success message."""
try:
from rich import print as rprint
rprint(f"[green]✓ {message}[/green]")
except ImportError:
print(f"✓ {message}")
def cmd_serve(self, args: List[str]) -> int:
"""Start the MCP server."""
parser = argparse.ArgumentParser(prog="praisonai mcp serve")
parser.add_argument("--transport", default="stdio", choices=["stdio", "http-stream"])
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8080)
parser.add_argument("--endpoint", default="/mcp")
parser.add_argument("--api-key", default=None)
parser.add_argument("--name", default="praisonai")
parser.add_argument("--debug", action="store_true")
parser.add_argument("--response-mode", default="batch", choices=["batch", "stream"])
parser.add_argument("--cors-origins", default=None, help="Comma-separated CORS origins")
parser.add_argument("--allowed-origins", default=None, help="Comma-separated allowed origins for security")
parser.add_argument("--session-ttl", type=int, default=3600, help="Session TTL in seconds")
parser.add_argument("--allow-termination", action="store_true", default=True, help="Allow client session termination")
parser.add_argument("--no-termination", action="store_true", help="Disable client session termination")
parser.add_argument("--resumability", action="store_true", default=True, help="Enable SSE resumability")
parser.add_argument("--log-level", default="warning", choices=["debug", "info", "warning", "error"])
parser.add_argument("--json", action="store_true", help="Output in JSON format")
try:
parsed = parser.parse_args(args)
except SystemExit:
return self.EXIT_ERROR
# Configure logging
log_levels = {
"debug": logging.DEBUG,
"info": logging.INFO,
"warning": logging.WARNING,
"error": logging.ERROR,
}
log_level = log_levels.get(parsed.log_level, logging.WARNING)
if parsed.debug:
log_level = logging.DEBUG
logging.basicConfig(level=log_level, stream=sys.stderr)
try:
from .server import MCPServer
from .adapters import register_all
# Register all tools, resources, and prompts
register_all()
# Create server
server = MCPServer(
name=parsed.name,
instructions="PraisonAI MCP Server - AI agent capabilities exposed via MCP protocol.",
)
if parsed.transport == "stdio":
# STDIO mode - minimal output to stderr
logger.info(f"Starting MCP server '{parsed.name}' on STDIO transport")
server.run_stdio()
else:
# HTTP Stream mode
cors_origins = parsed.cors_origins.split(",") if parsed.cors_origins else None
allowed_origins = parsed.allowed_origins.split(",") if parsed.allowed_origins else None
allow_termination = not parsed.no_termination
if not parsed.json:
self._print_success(f"Starting MCP server '{parsed.name}' on http://{parsed.host}:{parsed.port}{parsed.endpoint}")
server.run_http_stream(
host=parsed.host,
port=parsed.port,
endpoint=parsed.endpoint,
api_key=parsed.api_key,
cors_origins=cors_origins,
allowed_origins=allowed_origins,
session_ttl=parsed.session_ttl,
allow_client_termination=allow_termination,
response_mode=parsed.response_mode,
resumability_enabled=parsed.resumability,
)
return self.EXIT_SUCCESS
except ImportError as e:
self._print_error(f"Missing dependency: {e}")
print("Install with: pip install praisonai[mcp]", file=sys.stderr)
return self.EXIT_ERROR
except KeyboardInterrupt:
logger.info("Server stopped by user")
return self.EXIT_SUCCESS
except Exception as e:
self._print_error(str(e))
return self.EXIT_ERROR
def cmd_list_tools(self, args: List[str]) -> int:
"""List available MCP tools."""
try:
from .adapters import register_all_tools
from .registry import get_tool_registry
# Register all tools
register_all_tools()
registry = get_tool_registry()
tools = registry.list_schemas()
if not tools:
print("No tools registered")
return self.EXIT_SUCCESS
print(f"\n[bold]Available MCP Tools ({len(tools)}):[/bold]\n")
for tool in tools:
name = tool.get("name", "unknown")
desc = tool.get("description", "No description")
print(f" • {name}")
print(f" {desc}\n")
return self.EXIT_SUCCESS
except Exception as e:
self._print_error(str(e))
return self.EXIT_ERROR
def cmd_list_resources(self, args: List[str]) -> int:
"""List available MCP resources."""
try:
from .registry import get_resource_registry
registry = get_resource_registry()
resources = registry.list_schemas()
if not resources:
print("No resources registered")
return self.EXIT_SUCCESS
print(f"\n[bold]Available MCP Resources ({len(resources)}):[/bold]\n")
for res in resources:
uri = res.get("uri", "unknown")
desc = res.get("description", "No description")
print(f" • {uri}")
print(f" {desc}\n")
return self.EXIT_SUCCESS
except Exception as e:
self._print_error(str(e))
return self.EXIT_ERROR
def cmd_list_prompts(self, args: List[str]) -> int:
"""List available MCP prompts."""
try:
from .registry import get_prompt_registry
registry = get_prompt_registry()
prompts = registry.list_schemas()
if not prompts:
print("No prompts registered")
return self.EXIT_SUCCESS
print(f"\n[bold]Available MCP Prompts ({len(prompts)}):[/bold]\n")
for prompt in prompts:
name = prompt.get("name", "unknown")
desc = prompt.get("description", "No description")
print(f" • {name}")
print(f" {desc}\n")
return self.EXIT_SUCCESS
except Exception as e:
self._print_error(str(e))
return self.EXIT_ERROR
def cmd_config_generate(self, args: List[str]) -> int:
"""Generate client configuration."""
parser = argparse.ArgumentParser(prog="praisonai mcp config-generate")
parser.add_argument("--client", default="claude-desktop",
choices=["claude-desktop", "cursor", "vscode", "windsurf"])
parser.add_argument("--output", default=None)
parser.add_argument("--transport", default="stdio", choices=["stdio", "http-stream"])
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8080)
try:
parsed = parser.parse_args(args)
except SystemExit:
return self.EXIT_ERROR
# Generate config based on client type
if parsed.client == "claude-desktop":
config = self._generate_claude_desktop_config(parsed)
elif parsed.client == "cursor":
config = self._generate_cursor_config(parsed)
elif parsed.client == "vscode":
config = self._generate_vscode_config(parsed)
elif parsed.client == "windsurf":
config = self._generate_windsurf_config(parsed)
else:
config = self._generate_claude_desktop_config(parsed)
# Output config
config_json = json.dumps(config, indent=2)
if parsed.output:
with open(parsed.output, 'w') as f:
f.write(config_json)
self._print_success(f"Config written to {parsed.output}")
else:
print(f"\n[bold]{parsed.client} Configuration:[/bold]\n")
print(config_json)
print()
return self.EXIT_SUCCESS
def _generate_claude_desktop_config(self, args) -> dict:
"""Generate Claude Desktop config."""
if args.transport == "stdio":
return {
"mcpServers": {
"praisonai": {
"command": "praisonai",
"args": ["mcp", "serve", "--transport", "stdio"],
}
}
}
else:
return {
"mcpServers": {
"praisonai": {
"url": f"http://{args.host}:{args.port}/mcp",
"transport": "http-stream",
}
}
}
def _generate_cursor_config(self, args) -> dict:
"""Generate Cursor config."""
if args.transport == "stdio":
return {
"mcpServers": {
"praisonai": {
"command": "praisonai",
"args": ["mcp", "serve", "--transport", "stdio"],
}
}
}
else:
return {
"mcpServers": {
"praisonai": {
"url": f"http://{args.host}:{args.port}/mcp",
}
}
}
def _generate_vscode_config(self, args) -> dict:
"""Generate VSCode MCP config."""
if args.transport == "stdio":
return {
"mcp.servers": {
"praisonai": {
"command": "praisonai",
"args": ["mcp", "serve", "--transport", "stdio"],
}
}
}
else:
return {
"mcp.servers": {
"praisonai": {
"url": f"http://{args.host}:{args.port}/mcp",
}
}
}
def _generate_windsurf_config(self, args) -> dict:
"""Generate Windsurf config."""
return self._generate_cursor_config(args)
def cmd_doctor(self, args: List[str]) -> int:
"""Check MCP server health and configuration."""
try:
from .server import PROTOCOL_VERSION, SUPPORTED_VERSIONS
from .registry import get_tool_registry, get_resource_registry, get_prompt_registry
from .adapters import register_all
print("\n[bold cyan]PraisonAI MCP Server Health Check[/bold cyan]\n")
# Register all components
register_all()
# Check protocol version
print(f"[bold]Protocol Version:[/bold] {PROTOCOL_VERSION}")
print(f"[bold]Supported Versions:[/bold] {', '.join(SUPPORTED_VERSIONS)}")
print()
# Check registries
tools = get_tool_registry().list_all()
resources = get_resource_registry().list_all()
prompts = get_prompt_registry().list_all()
print("[bold]Registered Components:[/bold]")
print(f" • Tools: {len(tools)}")
print(f" • Resources: {len(resources)}")
print(f" • Prompts: {len(prompts)}")
print()
# Check environment
import os
env_checks = {
"OPENAI_API_KEY": bool(os.environ.get("OPENAI_API_KEY")),
"ANTHROPIC_API_KEY": bool(os.environ.get("ANTHROPIC_API_KEY")),
"GOOGLE_API_KEY": bool(os.environ.get("GOOGLE_API_KEY")),
}
print("[bold]Environment:[/bold]")
for key, present in env_checks.items():
status = "[green]✓[/green]" if present else "[yellow]○[/yellow]"
print(f" {status} {key}")
print()
# Check dependencies
deps = {
"starlette": False,
"uvicorn": False,
"praisonaiagents": False,
}
for dep in deps:
try:
__import__(dep)
deps[dep] = True
except ImportError:
pass
print("[bold]Dependencies:[/bold]")
for dep, available in deps.items():
status = "[green]✓[/green]" if available else "[red]✗[/red]"
print(f" {status} {dep}")
print()
# Overall status
all_deps = all(deps.values())
any_api_key = any(env_checks.values())
if all_deps and any_api_key:
self._print_success("MCP server is ready to run")
elif not all_deps:
self._print_error("Missing required dependencies. Install with: pip install praisonai[mcp]")
return self.EXIT_ERROR
else:
print("[yellow]Warning: No API keys configured. Some tools may not work.[/yellow]")
return self.EXIT_SUCCESS
except Exception as e:
self._print_error(str(e))
return self.EXIT_ERROR
def handle_mcp_command(args: List[str]) -> int:
"""
Entry point for MCP CLI commands.
Args:
args: Command arguments
Returns:
Exit code
"""
cli = MCPServerCLI()
return cli.handle(args)
@@ -0,0 +1,451 @@
"""
MCP Tool/Resource/Prompt Registry
Provides a centralized registry for MCP tools, resources, and prompts.
Supports lazy registration and schema generation.
"""
import inspect
import logging
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Union
logger = logging.getLogger(__name__)
@dataclass
class MCPToolDefinition:
"""Definition of an MCP tool."""
name: str
description: str
handler: Callable
input_schema: Dict[str, Any]
output_schema: Optional[Dict[str, Any]] = None
annotations: Optional[Dict[str, Any]] = None
def to_mcp_schema(self) -> Dict[str, Any]:
"""Convert to MCP tool schema format."""
schema = {
"name": self.name,
"description": self.description,
"inputSchema": self.input_schema,
}
if self.output_schema:
schema["outputSchema"] = self.output_schema
if self.annotations:
schema["annotations"] = self.annotations
return schema
@dataclass
class MCPResourceDefinition:
"""Definition of an MCP resource."""
uri: str
name: str
description: str
handler: Callable
mime_type: str = "application/json"
def to_mcp_schema(self) -> Dict[str, Any]:
"""Convert to MCP resource schema format."""
return {
"uri": self.uri,
"name": self.name,
"description": self.description,
"mimeType": self.mime_type,
}
@dataclass
class MCPPromptDefinition:
"""Definition of an MCP prompt."""
name: str
description: str
handler: Callable
arguments: List[Dict[str, Any]] = field(default_factory=list)
def to_mcp_schema(self) -> Dict[str, Any]:
"""Convert to MCP prompt schema format."""
return {
"name": self.name,
"description": self.description,
"arguments": self.arguments,
}
class MCPToolRegistry:
"""Registry for MCP tools."""
def __init__(self):
self._tools: Dict[str, MCPToolDefinition] = {}
self._lazy_loaders: List[Callable[[], None]] = []
def register(
self,
name: str,
handler: Callable,
description: Optional[str] = None,
input_schema: Optional[Dict[str, Any]] = None,
output_schema: Optional[Dict[str, Any]] = None,
annotations: Optional[Dict[str, Any]] = None,
) -> None:
"""Register a tool."""
if description is None:
description = inspect.getdoc(handler) or f"Tool: {name}"
if "\n" in description:
description = description.split("\n")[0]
if input_schema is None:
input_schema = self._generate_input_schema(handler)
self._tools[name] = MCPToolDefinition(
name=name,
description=description,
handler=handler,
input_schema=input_schema,
output_schema=output_schema,
annotations=annotations,
)
logger.debug(f"Registered MCP tool: {name}")
def register_lazy(self, loader: Callable[[], None]) -> None:
"""Register a lazy loader that will be called before listing tools."""
self._lazy_loaders.append(loader)
def _ensure_loaded(self) -> None:
"""Ensure all lazy loaders have been called."""
for loader in self._lazy_loaders:
try:
loader()
except Exception as e:
logger.warning(f"Lazy loader failed: {e}")
self._lazy_loaders.clear()
def get(self, name: str) -> Optional[MCPToolDefinition]:
"""Get a tool by name."""
self._ensure_loaded()
return self._tools.get(name)
def list_all(self) -> List[MCPToolDefinition]:
"""List all registered tools."""
self._ensure_loaded()
return list(self._tools.values())
def list_schemas(self) -> List[Dict[str, Any]]:
"""List all tool schemas."""
return [tool.to_mcp_schema() for tool in self.list_all()]
def _generate_input_schema(self, handler: Callable) -> Dict[str, Any]:
"""Generate JSON schema from function signature."""
sig = inspect.signature(handler)
hints = getattr(handler, "__annotations__", {})
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name in ("self", "cls"):
continue
prop = {"type": "string"} # Default
# Get type hint
hint = hints.get(param_name)
if hint:
prop = self._type_to_json_schema(hint)
# Get description from docstring if available
docstring = inspect.getdoc(handler)
if docstring:
# Try to extract param description from docstring
for line in docstring.split("\n"):
if f":param {param_name}:" in line:
# Extract description after the param marker
parts = line.split(f":param {param_name}:", 1)
if len(parts) > 1:
prop["description"] = parts[1].strip()
properties[param_name] = prop
if param.default is inspect.Parameter.empty:
required.append(param_name)
return {
"type": "object",
"properties": properties,
"required": required if required else None,
}
def _type_to_json_schema(self, hint: Any) -> Dict[str, Any]:
"""Convert Python type hint to JSON schema."""
if hint is str:
return {"type": "string"}
elif hint is int:
return {"type": "integer"}
elif hint is float:
return {"type": "number"}
elif hint is bool:
return {"type": "boolean"}
elif hint is list or (hasattr(hint, "__origin__") and hint.__origin__ is list):
return {"type": "array", "items": {"type": "string"}}
elif hint is dict or (hasattr(hint, "__origin__") and hint.__origin__ is dict):
return {"type": "object"}
else:
return {"type": "string"}
class MCPResourceRegistry:
"""Registry for MCP resources."""
def __init__(self):
self._resources: Dict[str, MCPResourceDefinition] = {}
self._lazy_loaders: List[Callable[[], None]] = []
def register(
self,
uri: str,
handler: Callable,
name: Optional[str] = None,
description: Optional[str] = None,
mime_type: str = "application/json",
) -> None:
"""Register a resource."""
if name is None:
name = uri.split("/")[-1] or uri
if description is None:
description = inspect.getdoc(handler) or f"Resource: {name}"
if "\n" in description:
description = description.split("\n")[0]
self._resources[uri] = MCPResourceDefinition(
uri=uri,
name=name,
description=description,
handler=handler,
mime_type=mime_type,
)
logger.debug(f"Registered MCP resource: {uri}")
def register_lazy(self, loader: Callable[[], None]) -> None:
"""Register a lazy loader."""
self._lazy_loaders.append(loader)
def _ensure_loaded(self) -> None:
"""Ensure all lazy loaders have been called."""
for loader in self._lazy_loaders:
try:
loader()
except Exception as e:
logger.warning(f"Lazy loader failed: {e}")
self._lazy_loaders.clear()
def get(self, uri: str) -> Optional[MCPResourceDefinition]:
"""Get a resource by URI."""
self._ensure_loaded()
return self._resources.get(uri)
def list_all(self) -> List[MCPResourceDefinition]:
"""List all registered resources."""
self._ensure_loaded()
return list(self._resources.values())
def list_schemas(self) -> List[Dict[str, Any]]:
"""List all resource schemas."""
return [res.to_mcp_schema() for res in self.list_all()]
class MCPPromptRegistry:
"""Registry for MCP prompts."""
def __init__(self):
self._prompts: Dict[str, MCPPromptDefinition] = {}
self._lazy_loaders: List[Callable[[], None]] = []
def register(
self,
name: str,
handler: Callable,
description: Optional[str] = None,
arguments: Optional[List[Dict[str, Any]]] = None,
) -> None:
"""Register a prompt."""
if description is None:
description = inspect.getdoc(handler) or f"Prompt: {name}"
if "\n" in description:
description = description.split("\n")[0]
if arguments is None:
arguments = self._generate_arguments(handler)
self._prompts[name] = MCPPromptDefinition(
name=name,
description=description,
handler=handler,
arguments=arguments,
)
logger.debug(f"Registered MCP prompt: {name}")
def register_lazy(self, loader: Callable[[], None]) -> None:
"""Register a lazy loader."""
self._lazy_loaders.append(loader)
def _ensure_loaded(self) -> None:
"""Ensure all lazy loaders have been called."""
for loader in self._lazy_loaders:
try:
loader()
except Exception as e:
logger.warning(f"Lazy loader failed: {e}")
self._lazy_loaders.clear()
def get(self, name: str) -> Optional[MCPPromptDefinition]:
"""Get a prompt by name."""
self._ensure_loaded()
return self._prompts.get(name)
def list_all(self) -> List[MCPPromptDefinition]:
"""List all registered prompts."""
self._ensure_loaded()
return list(self._prompts.values())
def list_schemas(self) -> List[Dict[str, Any]]:
"""List all prompt schemas."""
return [prompt.to_mcp_schema() for prompt in self.list_all()]
def _generate_arguments(self, handler: Callable) -> List[Dict[str, Any]]:
"""Generate prompt arguments from function signature."""
sig = inspect.signature(handler)
arguments = []
for param_name, param in sig.parameters.items():
if param_name in ("self", "cls"):
continue
arg = {
"name": param_name,
"required": param.default is inspect.Parameter.empty,
}
# Add description if available
docstring = inspect.getdoc(handler)
if docstring and f":param {param_name}:" in docstring:
# Extract description
pass
arguments.append(arg)
return arguments
# Global registries
_tool_registry = MCPToolRegistry()
_resource_registry = MCPResourceRegistry()
_prompt_registry = MCPPromptRegistry()
def register_tool(
name: str,
handler: Optional[Callable] = None,
description: Optional[str] = None,
input_schema: Optional[Dict[str, Any]] = None,
output_schema: Optional[Dict[str, Any]] = None,
annotations: Optional[Dict[str, Any]] = None,
) -> Union[Callable, None]:
"""
Register a tool with the global registry.
Can be used as a decorator or called directly.
Example:
@register_tool("praisonai.memory.show")
def show_memory(session_id: str = None) -> str:
'''Show memory contents.'''
...
# Or directly:
register_tool("praisonai.memory.show", show_memory)
"""
def decorator(fn: Callable) -> Callable:
_tool_registry.register(
name=name,
handler=fn,
description=description,
input_schema=input_schema,
output_schema=output_schema,
annotations=annotations,
)
return fn
if handler is not None:
decorator(handler)
return None
return decorator
def register_resource(
uri: str,
handler: Optional[Callable] = None,
name: Optional[str] = None,
description: Optional[str] = None,
mime_type: str = "application/json",
) -> Union[Callable, None]:
"""
Register a resource with the global registry.
Can be used as a decorator or called directly.
"""
def decorator(fn: Callable) -> Callable:
_resource_registry.register(
uri=uri,
handler=fn,
name=name,
description=description,
mime_type=mime_type,
)
return fn
if handler is not None:
decorator(handler)
return None
return decorator
def register_prompt(
name: str,
handler: Optional[Callable] = None,
description: Optional[str] = None,
arguments: Optional[List[Dict[str, Any]]] = None,
) -> Union[Callable, None]:
"""
Register a prompt with the global registry.
Can be used as a decorator or called directly.
"""
def decorator(fn: Callable) -> Callable:
_prompt_registry.register(
name=name,
handler=fn,
description=description,
arguments=arguments,
)
return fn
if handler is not None:
decorator(handler)
return None
return decorator
def get_tool_registry() -> MCPToolRegistry:
"""Get the global tool registry."""
return _tool_registry
def get_resource_registry() -> MCPResourceRegistry:
"""Get the global resource registry."""
return _resource_registry
def get_prompt_registry() -> MCPPromptRegistry:
"""Get the global prompt registry."""
return _prompt_registry
@@ -0,0 +1,427 @@
"""
MCP Server Implementation
Core MCP server that handles JSON-RPC messages and routes them to appropriate handlers.
Supports both STDIO and HTTP Stream transports.
"""
import asyncio
import json
import logging
from typing import Any, Callable, Dict, Optional, Set
from .registry import (
get_tool_registry,
get_resource_registry,
get_prompt_registry,
MCPToolRegistry,
MCPResourceRegistry,
MCPPromptRegistry,
)
logger = logging.getLogger(__name__)
# MCP Protocol Constants (Updated to 2025-11-25 spec)
PROTOCOL_VERSION = "2025-11-25"
SUPPORTED_VERSIONS = ["2025-11-25", "2025-03-26", "2024-11-05"]
# JSON-RPC Error Codes
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
class MCPServer:
"""
MCP Server that exposes PraisonAI capabilities.
Supports:
- STDIO transport (for Claude Desktop, Cursor, etc.)
- HTTP Stream transport (MCP 2025-11-25 spec)
Example:
server = MCPServer(name="praisonai")
server.run(transport="stdio")
"""
def __init__(
self,
name: str = "praisonai",
version: str = "1.0.0",
tool_registry: Optional[MCPToolRegistry] = None,
resource_registry: Optional[MCPResourceRegistry] = None,
prompt_registry: Optional[MCPPromptRegistry] = None,
instructions: Optional[str] = None,
):
"""
Initialize MCP server.
Args:
name: Server name
version: Server version
tool_registry: Custom tool registry (uses global if None)
resource_registry: Custom resource registry (uses global if None)
prompt_registry: Custom prompt registry (uses global if None)
instructions: Optional instructions for clients
"""
self.name = name
self.version = version
self.instructions = instructions
self._tool_registry = tool_registry or get_tool_registry()
self._resource_registry = resource_registry or get_resource_registry()
self._prompt_registry = prompt_registry or get_prompt_registry()
self._initialized = False
self._client_info: Optional[Dict[str, Any]] = None
self._protocol_version: str = PROTOCOL_VERSION
# Cancellation support
self._active_requests: Dict[str, asyncio.Task] = {}
self._cancelled_requests: Set[str] = set()
# Progress notification callback
self._progress_callback: Optional[Callable] = None
# Method handlers
self._handlers: Dict[str, Callable] = {
"initialize": self._handle_initialize,
"ping": self._handle_ping,
"tools/list": self._handle_tools_list,
"tools/call": self._handle_tools_call,
"resources/list": self._handle_resources_list,
"resources/read": self._handle_resources_read,
"prompts/list": self._handle_prompts_list,
"prompts/get": self._handle_prompts_get,
"logging/setLevel": self._handle_set_log_level,
}
def get_capabilities(self) -> Dict[str, Any]:
"""Get server capabilities per MCP 2025-11-25 spec."""
return {
"tools": {
"listChanged": True,
},
"resources": {
"subscribe": False,
"listChanged": True,
},
"prompts": {
"listChanged": True,
},
"logging": {},
}
def set_progress_callback(self, callback: Callable) -> None:
"""Set callback for progress notifications."""
self._progress_callback = callback
async def send_progress(self, progress_token: str, progress: float, total: Optional[float] = None) -> None:
"""Send progress notification for long-running operations."""
if self._progress_callback:
notification = {
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": {
"progressToken": progress_token,
"progress": progress,
}
}
if total is not None:
notification["params"]["total"] = total
await self._progress_callback(notification)
def is_cancelled(self, request_id: str) -> bool:
"""Check if a request has been cancelled."""
return request_id in self._cancelled_requests
async def handle_message(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Handle an incoming JSON-RPC message.
Args:
message: JSON-RPC message
Returns:
Response message or None for notifications
"""
# Validate JSON-RPC format
if message.get("jsonrpc") != "2.0":
return self._error_response(None, INVALID_REQUEST, "Invalid JSON-RPC version")
method = message.get("method")
params = message.get("params", {})
msg_id = message.get("id")
# Notifications don't have an id
is_notification = msg_id is None
# Handle initialized notification
if method == "notifications/initialized":
self._initialized = True
logger.debug("Client sent initialized notification")
return None
# Handle cancellation notification
if method == "notifications/cancelled":
request_id = params.get("requestId")
if request_id:
self._cancelled_requests.add(str(request_id))
# Cancel active task if exists
if str(request_id) in self._active_requests:
task = self._active_requests[str(request_id)]
task.cancel()
logger.debug(f"Cancelled request: {request_id}")
return None
# Check if method exists
handler = self._handlers.get(method)
if handler is None:
if is_notification:
return None
return self._error_response(msg_id, METHOD_NOT_FOUND, f"Method not found: {method}")
# Execute handler
try:
result = await handler(params)
if is_notification:
return None
return self._success_response(msg_id, result)
except Exception as e:
logger.exception(f"Error handling {method}")
if is_notification:
return None
return self._error_response(msg_id, INTERNAL_ERROR, str(e))
async def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle initialize request."""
self._client_info = params.get("clientInfo")
client_version = params.get("protocolVersion", PROTOCOL_VERSION)
# Negotiate protocol version
if client_version in SUPPORTED_VERSIONS:
self._protocol_version = client_version
else:
self._protocol_version = PROTOCOL_VERSION
logger.info(f"Client initialized: {self._client_info}, protocol: {self._protocol_version}")
result = {
"protocolVersion": self._protocol_version,
"capabilities": self.get_capabilities(),
"serverInfo": {
"name": self.name,
"version": self.version,
},
}
if self.instructions:
result["instructions"] = self.instructions
return result
async def _handle_ping(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle ping request."""
return {}
async def _handle_tools_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle tools/list request."""
tools = self._tool_registry.list_schemas()
return {"tools": tools}
async def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle tools/call request."""
tool_name = params.get("name")
arguments = params.get("arguments", {})
if not tool_name:
raise ValueError("Tool name required")
tool = self._tool_registry.get(tool_name)
if tool is None:
raise ValueError(f"Tool not found: {tool_name}")
# Execute tool
try:
if asyncio.iscoroutinefunction(tool.handler):
result = await tool.handler(**arguments)
else:
result = tool.handler(**arguments)
# Format result as MCP content
if isinstance(result, str):
content = [{"type": "text", "text": result}]
elif isinstance(result, dict):
content = [{"type": "text", "text": json.dumps(result, indent=2)}]
elif isinstance(result, list):
content = [{"type": "text", "text": json.dumps(result, indent=2)}]
else:
content = [{"type": "text", "text": str(result)}]
return {"content": content, "isError": False}
except Exception as e:
logger.exception(f"Tool execution error: {tool_name}")
return {
"content": [{"type": "text", "text": f"Error: {str(e)}"}],
"isError": True,
}
async def _handle_resources_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle resources/list request."""
resources = self._resource_registry.list_schemas()
return {"resources": resources}
async def _handle_resources_read(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle resources/read request."""
uri = params.get("uri")
if not uri:
raise ValueError("Resource URI required")
resource = self._resource_registry.get(uri)
if resource is None:
raise ValueError(f"Resource not found: {uri}")
# Execute resource handler
try:
if asyncio.iscoroutinefunction(resource.handler):
result = await resource.handler()
else:
result = resource.handler()
# Format result
if isinstance(result, str):
content = [{"uri": uri, "mimeType": resource.mime_type, "text": result}]
elif isinstance(result, dict):
content = [{"uri": uri, "mimeType": "application/json", "text": json.dumps(result)}]
else:
content = [{"uri": uri, "mimeType": resource.mime_type, "text": str(result)}]
return {"contents": content}
except Exception:
logger.exception(f"Resource read error: {uri}")
raise
async def _handle_prompts_list(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle prompts/list request."""
prompts = self._prompt_registry.list_schemas()
return {"prompts": prompts}
async def _handle_prompts_get(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle prompts/get request."""
prompt_name = params.get("name")
arguments = params.get("arguments", {})
if not prompt_name:
raise ValueError("Prompt name required")
prompt = self._prompt_registry.get(prompt_name)
if prompt is None:
raise ValueError(f"Prompt not found: {prompt_name}")
# Execute prompt handler
try:
if asyncio.iscoroutinefunction(prompt.handler):
result = await prompt.handler(**arguments)
else:
result = prompt.handler(**arguments)
# Format result as messages
if isinstance(result, str):
messages = [{"role": "user", "content": {"type": "text", "text": result}}]
elif isinstance(result, list):
messages = result
else:
messages = [{"role": "user", "content": {"type": "text", "text": str(result)}}]
return {"messages": messages}
except Exception:
logger.exception(f"Prompt get error: {prompt_name}")
raise
async def _handle_set_log_level(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle logging/setLevel request per MCP spec."""
level = params.get("level", "info")
level_map = {
"debug": logging.DEBUG,
"info": logging.INFO,
"notice": logging.INFO,
"warning": logging.WARNING,
"error": logging.ERROR,
"critical": logging.CRITICAL,
"alert": logging.CRITICAL,
"emergency": logging.CRITICAL,
}
log_level = level_map.get(level.lower(), logging.INFO)
logging.getLogger().setLevel(log_level)
logger.info(f"Log level set to: {level}")
return {}
def _success_response(self, msg_id: Any, result: Any) -> Dict[str, Any]:
"""Create a success response."""
return {
"jsonrpc": "2.0",
"id": msg_id,
"result": result,
}
def _error_response(self, msg_id: Any, code: int, message: str, data: Any = None) -> Dict[str, Any]:
"""Create an error response."""
error = {"code": code, "message": message}
if data is not None:
error["data"] = data
return {
"jsonrpc": "2.0",
"id": msg_id,
"error": error,
}
def run(self, transport: str = "stdio", **kwargs) -> None:
"""
Run the MCP server.
Args:
transport: Transport type ("stdio" or "http-stream")
**kwargs: Transport-specific options
"""
if transport == "stdio":
self.run_stdio()
elif transport == "http-stream":
self.run_http_stream(**kwargs)
else:
raise ValueError(f"Unknown transport: {transport}")
def run_stdio(self) -> None:
"""Run server with STDIO transport."""
from .transports.stdio import StdioTransport
transport = StdioTransport(self)
asyncio.run(transport.run())
def run_http_stream(
self,
host: str = "127.0.0.1",
port: int = 8080,
endpoint: str = "/mcp",
**kwargs,
) -> None:
"""
Run server with HTTP Stream transport.
Args:
host: Server host
port: Server port
endpoint: MCP endpoint path
"""
from .transports.http_stream import HTTPStreamTransport
transport = HTTPStreamTransport(
server=self,
host=host,
port=port,
endpoint=endpoint,
**kwargs,
)
transport.run()
@@ -0,0 +1,18 @@
"""
MCP Transport Implementations
Provides STDIO and HTTP Stream transports for the MCP server.
"""
__all__ = ["StdioTransport", "HTTPStreamTransport"]
def __getattr__(name):
"""Lazy load transports."""
if name == "StdioTransport":
from .stdio import StdioTransport
return StdioTransport
elif name == "HTTPStreamTransport":
from .http_stream import HTTPStreamTransport
return HTTPStreamTransport
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,376 @@
"""
HTTP Stream Transport for MCP Server
Implements the MCP Streamable HTTP transport (MCP 2025-11-25 spec):
- Single endpoint for all MCP communication
- POST for client→server messages
- GET for server→client SSE stream (optional)
- Session management via MCP-Session-Id header
- Supports batch (JSON) and stream (SSE) response modes
- Origin header validation for security
- MCP-Protocol-Version header support
"""
import asyncio
import json
import logging
import time
import uuid
from typing import TYPE_CHECKING, Any, Dict, Optional
if TYPE_CHECKING:
from ..server import MCPServer
logger = logging.getLogger(__name__)
# MCP Protocol Version (Updated to 2025-11-25)
PROTOCOL_VERSION = "2025-11-25"
SUPPORTED_VERSIONS = ["2025-11-25", "2025-03-26"]
class HTTPStreamTransport:
"""
HTTP Stream transport for MCP server.
Implements the Streamable HTTP transport per MCP 2025-11-25 spec.
"""
def __init__(
self,
server: "MCPServer",
host: str = "127.0.0.1",
port: int = 8080,
endpoint: str = "/mcp",
cors_origins: Optional[list] = None,
allowed_origins: Optional[list] = None,
api_key: Optional[str] = None,
session_ttl: int = 3600,
allow_client_termination: bool = True,
response_mode: str = "batch",
resumability_enabled: bool = True,
event_history_duration: int = 300,
):
"""
Initialize HTTP Stream transport.
Args:
server: MCPServer instance
host: Server host
port: Server port
endpoint: MCP endpoint path
cors_origins: CORS allowed origins
allowed_origins: Origins allowed for security validation (None = localhost only)
api_key: Optional API key for authentication
session_ttl: Session TTL in seconds
allow_client_termination: Allow client to terminate session via DELETE
response_mode: Default response mode ("batch" or "stream")
resumability_enabled: Enable SSE resumability
event_history_duration: Duration to keep event history (seconds)
"""
self.server = server
self.host = host
self.port = port
self.endpoint = endpoint
self.cors_origins = cors_origins or ["*"]
self.api_key = api_key
self.session_ttl = session_ttl
self.allow_client_termination = allow_client_termination
self.response_mode = response_mode
self.resumability_enabled = resumability_enabled
self.event_history_duration = event_history_duration
# Allowed origins for security (MCP 2025-11-25 requirement)
# Default: only localhost if binding to localhost
if allowed_origins is None:
if host in ("127.0.0.1", "localhost", "::1"):
self.allowed_origins = ["http://localhost", "http://127.0.0.1", "https://localhost", "https://127.0.0.1"]
else:
# External binding requires explicit origin config
self.allowed_origins = None # Will reject all Origin headers
logger.warning("External binding without allowed_origins - Origin validation will reject all requests with Origin header")
else:
self.allowed_origins = allowed_origins
# Session storage
self._sessions: Dict[str, Dict[str, Any]] = {}
# SSE event history for resumability
self._event_history: Dict[str, list] = {}
self._event_counter = 0
self._event_timestamps: Dict[str, float] = {}
def run(self) -> None:
"""Run the HTTP Stream server."""
try:
import uvicorn
except ImportError:
raise ImportError("uvicorn required. Install with: pip install uvicorn")
app = self._create_app()
logger.info(f"MCP server '{self.server.name}' starting on http://{self.host}:{self.port}{self.endpoint}")
uvicorn.run(app, host=self.host, port=self.port, log_level="warning")
def _validate_origin(self, request_origin: Optional[str]) -> bool:
"""
Validate Origin header per MCP 2025-11-25 security requirements.
Returns True if origin is valid or not present.
Returns False if origin is present but invalid.
"""
if request_origin is None:
# No Origin header - allow (same-origin requests don't send Origin)
return True
if self.allowed_origins is None:
# No allowed origins configured - reject all Origin headers
return False
# Check if origin matches any allowed origin
for allowed in self.allowed_origins:
if request_origin == allowed or request_origin.startswith(allowed):
return True
return False
def _validate_protocol_version(self, version: Optional[str]) -> bool:
"""Validate MCP-Protocol-Version header."""
if version is None:
# No version header - assume 2025-03-26 for backwards compatibility
return True
return version in SUPPORTED_VERSIONS
def _create_app(self) -> Any:
"""Create the Starlette/FastAPI application."""
try:
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse
from starlette.routing import Route
except ImportError:
raise ImportError("starlette required. Install with: pip install starlette")
async def mcp_post(request: Request) -> Response:
"""Handle POST requests (client→server messages)."""
# Validate Origin header (MCP 2025-11-25 security requirement)
origin = request.headers.get("Origin")
if not self._validate_origin(origin):
logger.warning(f"Rejected request with invalid Origin: {origin}")
return JSONResponse(
{"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Forbidden: Invalid Origin"}},
status_code=403,
)
# Validate MCP-Protocol-Version header
protocol_version = request.headers.get("MCP-Protocol-Version")
if protocol_version and not self._validate_protocol_version(protocol_version):
return JSONResponse(
{"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": f"Unsupported protocol version: {protocol_version}"}},
status_code=400,
)
# Check authentication
if self.api_key:
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer ") or auth_header[7:] != self.api_key:
return JSONResponse(
{"error": "Unauthorized"},
status_code=401,
)
# Get or create session (check both header casings for compatibility)
session_id = request.headers.get("MCP-Session-Id") or request.headers.get("Mcp-Session-Id")
if session_id and session_id not in self._sessions:
# Session expired
return JSONResponse(
{"error": "Session not found"},
status_code=404,
)
# Parse request body
try:
body = await request.json()
except json.JSONDecodeError:
return JSONResponse(
{"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}},
status_code=400,
)
# Handle message
response = await self.server.handle_message(body)
# Check if this is an initialize request
if body.get("method") == "initialize":
# Create new session
new_session_id = str(uuid.uuid4())
self._sessions[new_session_id] = {
"created_at": time.time(),
"last_activity": time.time(),
}
# Add session ID to response headers (MCP 2025-11-25 uses MCP-Session-Id)
headers = {
"MCP-Session-Id": new_session_id,
"MCP-Protocol-Version": PROTOCOL_VERSION,
}
if response:
return JSONResponse(response, headers=headers)
return Response(status_code=202, headers=headers)
# Update session activity
if session_id and session_id in self._sessions:
self._sessions[session_id]["last_activity"] = time.time()
# Check Accept header for response mode
accept = request.headers.get("Accept", "application/json")
if "text/event-stream" in accept and response:
# Stream mode - return SSE
async def generate_sse():
event_id = self._next_event_id()
data = json.dumps(response)
yield f"id:{event_id}\ndata:{data}\n\n"
return StreamingResponse(
generate_sse(),
media_type="text/event-stream",
headers={"MCP-Protocol-Version": PROTOCOL_VERSION},
)
# Batch mode - return JSON
if response:
return JSONResponse(
response,
headers={"MCP-Protocol-Version": PROTOCOL_VERSION},
)
return Response(
status_code=202,
headers={"MCP-Protocol-Version": PROTOCOL_VERSION},
)
async def mcp_get(request: Request) -> Response:
"""Handle GET requests (server→client SSE stream)."""
# Check both header casings for compatibility
session_id = request.headers.get("MCP-Session-Id") or request.headers.get("Mcp-Session-Id")
if not session_id or session_id not in self._sessions:
return JSONResponse(
{"error": "Session required"},
status_code=400,
)
# Check for Last-Event-ID for resumability
last_event_id = request.headers.get("Last-Event-ID")
async def generate_sse():
# If resuming, replay missed events
if last_event_id and session_id in self._event_history:
history = self._event_history[session_id]
replay = False
for event_id, data in history:
if replay:
yield f"id:{event_id}\ndata:{data}\n\n"
elif event_id == last_event_id:
replay = True
# Keep connection alive
while True:
await asyncio.sleep(30)
yield ":keepalive\n\n"
return StreamingResponse(
generate_sse(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"MCP-Protocol-Version": PROTOCOL_VERSION,
},
)
async def mcp_delete(request: Request) -> Response:
"""Handle DELETE requests (session termination)."""
if not self.allow_client_termination:
return Response(status_code=405)
session_id = request.headers.get("MCP-Session-Id") or request.headers.get("Mcp-Session-Id")
if session_id and session_id in self._sessions:
del self._sessions[session_id]
if session_id in self._event_history:
del self._event_history[session_id]
return Response(status_code=204)
return JSONResponse(
{"error": "Session not found"},
status_code=404,
)
async def mcp_options(request: Request) -> Response:
"""Handle OPTIONS requests for CORS preflight."""
return Response(
status_code=204,
headers={
"Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS",
"Access-Control-Allow-Headers": "Content-Type, Accept, Authorization, MCP-Session-Id, Mcp-Session-Id, MCP-Protocol-Version, Mcp-Protocol-Version, Last-Event-ID",
},
)
async def health(request: Request) -> Response:
"""Health check endpoint."""
return JSONResponse({
"status": "healthy",
"server": self.server.name,
"version": self.server.version,
"protocol_version": PROTOCOL_VERSION,
"active_sessions": len(self._sessions),
})
async def root(request: Request) -> Response:
"""Root endpoint."""
return JSONResponse({
"message": f"PraisonAI MCP Server: {self.server.name}",
"mcp_endpoint": self.endpoint,
"protocol_version": PROTOCOL_VERSION,
})
# Create routes
routes = [
Route(self.endpoint, mcp_post, methods=["POST"]),
Route(self.endpoint, mcp_get, methods=["GET"]),
Route(self.endpoint, mcp_delete, methods=["DELETE"]),
Route(self.endpoint, mcp_options, methods=["OPTIONS"]),
Route("/health", health, methods=["GET"]),
Route("/", root, methods=["GET"]),
]
# Create middleware
middleware = [
Middleware(
CORSMiddleware,
allow_origins=self.cors_origins,
allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
allow_headers=["*"],
),
]
return Starlette(routes=routes, middleware=middleware)
def _next_event_id(self) -> str:
"""Generate next SSE event ID."""
self._event_counter += 1
return str(self._event_counter)
def _cleanup_sessions(self) -> None:
"""Clean up expired sessions."""
now = time.time()
expired = [
sid for sid, data in self._sessions.items()
if now - data["last_activity"] > self.session_ttl
]
for sid in expired:
del self._sessions[sid]
if sid in self._event_history:
del self._event_history[sid]
@@ -0,0 +1,132 @@
"""
STDIO Transport for MCP Server
Implements the MCP STDIO transport protocol:
- JSON-RPC 2.0 messages over stdin/stdout
- Messages delimited by newlines
- Logs go to stderr only
"""
import asyncio
import json
import logging
import sys
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..server import MCPServer
logger = logging.getLogger(__name__)
class StdioTransport:
"""
STDIO transport for MCP server.
Reads JSON-RPC messages from stdin and writes responses to stdout.
All logging goes to stderr to avoid corrupting the protocol stream.
"""
def __init__(self, server: "MCPServer"):
"""
Initialize STDIO transport.
Args:
server: MCPServer instance to handle messages
"""
self.server = server
self._running = False
async def run(self) -> None:
"""Run the STDIO transport loop."""
self._running = True
# Configure logging to stderr only
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
logger.addHandler(handler)
logger.info(f"MCP server '{self.server.name}' starting on STDIO transport")
# Use asyncio streams for non-blocking I/O
loop = asyncio.get_event_loop()
reader = asyncio.StreamReader()
protocol = asyncio.StreamReaderProtocol(reader)
await loop.connect_read_pipe(lambda: protocol, sys.stdin)
# Create writer for stdout
write_transport, write_protocol = await loop.connect_write_pipe(
asyncio.streams.FlowControlMixin, sys.stdout
)
writer = asyncio.StreamWriter(write_transport, write_protocol, reader, loop)
try:
while self._running:
try:
# Read a line (JSON-RPC message)
line = await reader.readline()
if not line:
# EOF reached
break
line = line.decode("utf-8").strip()
if not line:
continue
# Parse JSON-RPC message
try:
message = json.loads(line)
except json.JSONDecodeError as e:
logger.error(f"JSON parse error: {e}")
error_response = {
"jsonrpc": "2.0",
"id": None,
"error": {
"code": -32700,
"message": f"Parse error: {e}",
},
}
await self._write_message(writer, error_response)
continue
# Handle message
response = await self.server.handle_message(message)
# Write response if not a notification
if response is not None:
await self._write_message(writer, response)
except asyncio.CancelledError:
break
except Exception as e:
logger.exception(f"Error processing message: {e}")
finally:
self._running = False
writer.close()
logger.info("MCP server stopped")
async def _write_message(self, writer: asyncio.StreamWriter, message: dict) -> None:
"""Write a JSON-RPC message to stdout."""
try:
data = json.dumps(message) + "\n"
writer.write(data.encode("utf-8"))
await writer.drain()
except Exception as e:
logger.error(f"Error writing message: {e}")
def stop(self) -> None:
"""Stop the transport."""
self._running = False
def run_stdio_server(server: "MCPServer") -> None:
"""
Convenience function to run a server with STDIO transport.
Args:
server: MCPServer instance
"""
transport = StdioTransport(server)
asyncio.run(transport.run())
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.7.6"
__version__ = "2.7.7"
+2 -1
View File
@@ -36,7 +36,8 @@ gradio = ["gradio>=4.26.0"]
api = [
"flask>=3.0.0",
"fastapi>=0.115.0",
"uvicorn>=0.34.0"
"uvicorn>=0.34.0",
"sse-starlette>=1.6.0"
]
agentops = ["agentops>=0.3.12"]
google = ["langchain-google-genai>=2.1.0"]
+31
View File
@@ -2,9 +2,40 @@ import pytest
import os
import sys
import asyncio
import warnings
import gc
from unittest.mock import Mock, patch
from typing import Dict, Any, List
# Register pytest-asyncio plugin
pytest_plugins = ('pytest_asyncio',)
# Suppress aiohttp unclosed session warnings during tests
warnings.filterwarnings("ignore", message="Unclosed client session")
warnings.filterwarnings("ignore", message="Unclosed connector")
@pytest.fixture(scope="session")
def event_loop_policy():
"""Use default event loop policy."""
return asyncio.DefaultEventLoopPolicy()
@pytest.fixture(autouse=True)
def cleanup_async_resources():
"""Clean up async resources after each test to prevent unclosed session warnings."""
yield
# Force garbage collection to clean up any lingering async resources
gc.collect()
# Give the event loop a chance to clean up
try:
loop = asyncio.get_event_loop()
if loop.is_running():
pass # Can't run cleanup if loop is running
elif not loop.is_closed():
# Run any pending callbacks
loop.run_until_complete(asyncio.sleep(0))
except RuntimeError:
pass # No event loop, nothing to clean up
# Add the source path to sys.path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src', 'praisonai-agents'))
+2
View File
@@ -3986,6 +3986,7 @@ anthropic = [
api = [
{ name = "fastapi" },
{ name = "flask" },
{ name = "sse-starlette" },
{ name = "uvicorn" },
]
autogen = [
@@ -4133,6 +4134,7 @@ requires-dist = [
{ name = "sqlalchemy", marker = "extra == 'code'", specifier = ">=2.0.36" },
{ name = "sqlalchemy", marker = "extra == 'realtime'", specifier = ">=2.0.36" },
{ name = "sqlalchemy", marker = "extra == 'ui'", specifier = ">=2.0.36" },
{ name = "sse-starlette", marker = "extra == 'api'", specifier = ">=1.6.0" },
{ name = "tavily-python", marker = "extra == 'chat'", specifier = "==0.5.0" },
{ name = "tavily-python", marker = "extra == 'code'", specifier = "==0.5.0" },
{ name = "tavily-python", marker = "extra == 'realtime'", specifier = "==0.5.0" },