"""
Agent Orchestration Platform - FastMCP Server Entry Point
This module provides the main entry point for the Agent Orchestration Platform MCP server,
implementing comprehensive multi-agent Claude Code orchestration with security-first design.
Author: Adder_5 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import asyncio
import sys
import logging
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, Optional
# Add project root to Python path for imports
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from fastmcp import FastMCP, Context
from fastmcp.server.auth import BearerAuthProvider
# Configure logging to stderr only (no print statements for MCP compliance)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(sys.stderr), # STDERR only for MCP compliance
logging.FileHandler("logs/server.log", mode="a"),
],
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP(
name="AgentOrchestrationPlatform",
instructions="Agent Orchestration Platform for managing multiple Claude Code agents with comprehensive task coordination, security, and monitoring capabilities."
)
@mcp.tool()
async def create_agent(
session_id: str,
agent_name: str,
specialization: Optional[str] = None,
system_prompt_suffix: str = "",
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
Create a new Claude Code agent within a managed session.
Creates a new agent instance with the specified configuration and assigns it
to the given session for collaborative development work.
"""
try:
if ctx:
await ctx.info(f"Creating agent {agent_name} in session {session_id}")
# Basic validation
if not agent_name.startswith("Agent_"):
return {
"success": False,
"error": "Agent name must start with 'Agent_'"
}
# Simulate agent creation (in real implementation, this would start a Claude Code process)
agent_id = agent_name
if ctx:
await ctx.info(f"Agent {agent_id} created successfully")
return {
"success": True,
"agent_id": agent_id,
"agent_name": agent_name,
"specialization": specialization or "general",
"session_id": session_id,
"status": "created"
}
except Exception as e:
logger.error(f"Failed to create agent: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def create_session(
root_path: str,
session_name: str,
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
Create a new development session for a project.
Initializes a new session with the specified root path and prepares
the environment for agent collaboration.
"""
try:
if ctx:
await ctx.info(f"Creating session {session_name} at {root_path}")
# Validate root path exists
path = Path(root_path)
if not path.exists():
return {
"success": False,
"error": f"Root path does not exist: {root_path}"
}
session_id = f"session_{session_name}"
if ctx:
await ctx.info(f"Session {session_id} created successfully")
return {
"success": True,
"session_id": session_id,
"session_name": session_name,
"root_path": str(path.absolute()),
"status": "active"
}
except Exception as e:
logger.error(f"Failed to create session: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def send_message_to_agent(
agent_name: str,
message: str,
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
Send a message to a specific agent.
Delivers a message to the specified agent for processing.
The agent will receive the message and can respond accordingly.
"""
try:
if ctx:
await ctx.info(f"Sending message to {agent_name}")
# Validate agent name
if not agent_name.startswith("Agent_"):
return {
"success": False,
"error": "Invalid agent name format"
}
# Simulate message delivery
if ctx:
await ctx.info(f"Message delivered to {agent_name}")
return {
"success": True,
"agent_name": agent_name,
"message_sent": True,
"timestamp": str(datetime.now())
}
except Exception as e:
logger.error(f"Failed to send message: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def get_session_status(
session_id: str,
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
Get the current status of a session and its agents.
Returns comprehensive status information about the session
including all active agents and their current state.
"""
try:
if ctx:
await ctx.info(f"Getting status for session {session_id}")
# Simulate session status retrieval
return {
"success": True,
"session_id": session_id,
"status": "active",
"agents": [
{
"name": "Agent_1",
"status": "idle",
"specialization": "general"
}
],
"active_tasks": 0,
"completed_tasks": 0
}
except Exception as e:
logger.error(f"Failed to get session status: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def delete_agent(
agent_name: str,
force: bool = False,
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
Delete an agent from the system.
Safely terminates and removes an agent from its session.
Use force=True to forcefully terminate unresponsive agents.
"""
try:
if ctx:
await ctx.info(f"Deleting agent {agent_name}")
# Validate agent name
if not agent_name.startswith("Agent_"):
return {
"success": False,
"error": "Invalid agent name format"
}
# Simulate agent deletion
if ctx:
await ctx.info(f"Agent {agent_name} deleted successfully")
return {
"success": True,
"agent_name": agent_name,
"deleted": True,
"forced": force
}
except Exception as e:
logger.error(f"Failed to delete agent: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def delete_session(
session_id: str,
force: bool = False,
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
Delete a session and all its agents.
Safely terminates all agents in the session and cleans up resources.
Use force=True to forcefully terminate unresponsive sessions.
"""
try:
if ctx:
await ctx.info(f"Deleting session {session_id}")
# Simulate session deletion
if ctx:
await ctx.info(f"Session {session_id} deleted successfully")
return {
"success": True,
"session_id": session_id,
"deleted": True,
"forced": force,
"agents_terminated": 1
}
except Exception as e:
logger.error(f"Failed to delete session: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.tool()
async def list_sessions(
ctx: Optional[Context] = None
) -> Dict[str, Any]:
"""
List all active sessions and their basic information.
Returns a summary of all currently active sessions in the system.
"""
try:
if ctx:
await ctx.info("Listing all sessions")
# Simulate session listing
return {
"success": True,
"sessions": [
{
"session_id": "session_example",
"session_name": "example",
"status": "active",
"agent_count": 1,
"root_path": "/example/path"
}
],
"total_sessions": 1
}
except Exception as e:
logger.error(f"Failed to list sessions: {e}")
return {
"success": False,
"error": str(e)
}
@mcp.resource("session://{session_id}/info")
async def get_session_info(session_id: str) -> Dict[str, Any]:
"""Get detailed information about a specific session."""
return {
"session_id": session_id,
"status": "active",
"created_at": "2025-06-27T00:00:00Z",
"agents": ["Agent_1"],
"root_path": "/example/path"
}
def main():
"""Main entry point for the MCP server."""
try:
# Ensure logs directory exists
Path("logs").mkdir(exist_ok=True)
logger.info("Starting Agent Orchestration Platform MCP Server")
# Run the FastMCP server with stdio transport (default for MCP)
mcp.run(transport="stdio")
except Exception as e:
logger.error(f"Server failed to start: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
