docs/DEVELOPER_GUIDE.md

EcoMCP Developer Guide

Quick Start

Installation

# Install dependencies
pip install -r requirements.txt

# Run the UI (automatically starts server)
python run_ui.py

# Open browser to http://localhost:7860

Project Structure

ecomcp/
 src/                          # Main application code
    ui/                       # User Interface Layer
       app.py               # Main Gradio application
       components.py        # Reusable UI components
    server/                  # MCP Server Layer
       mcp_server.py        # JSON-RPC MCP implementation
    clients/                 # Client Layer
       mcp_client.py        # Server communication
    core/                    # Shared Logic (future)

 run_ui.py                    # Start UI
 run_server.py                # Start server standalone
 archive/                     # Deprecated files
 tests/                       # Test suite (future)

Key Modules

src.ui.app

Main Gradio application. Define all UI tabs and interface here.

Key Components:

• create_app() - Main factory function
• create_theme() - Theme configuration
• Tab definitions for each tool

Example:

from src.ui.app import create_app
app = create_app()
app.launch()

src.ui.components

Reusable UI components and tool handlers.

Key Classes:

• ToolCallHandler - Handles all MCP tool calls with streaming
  • analyze_product()
  • analyze_reviews()
  • generate_listing()
  • price_recommendation()

Key Functions:

• create_sentiment_chart_html() - Generate sentiment visualization
• create_pricing_tiers_html() - Generate pricing visualization

Example:

from src.ui.components import ToolCallHandler
from src.clients.mcp_client import MCPClient

client = MCPClient()
handler = ToolCallHandler(client)

# Stream product analysis
async for chunk in handler.analyze_product("Headphones", "electronics"):
    print(chunk)

src.server.mcp_server

JSON-RPC 2.0 MCP Protocol implementation.

Key Classes:

• EcoMCPServer - Main server class
  • handle_initialize() - Initialize request
  • handle_list_tools() - List available tools
  • call_tool() - Execute a tool

Key Methods:

• _analyze_product() - Product analysis
• _analyze_reviews() - Review sentiment analysis
• _generate_listing() - Product copy generation
• _price_recommendation() - Pricing strategy

Example:

from src.server.mcp_server import EcoMCPServer

server = EcoMCPServer()
result = await server.call_tool("analyze_product", {
    "product_name": "Headphones",
    "category": "electronics"
})

src.clients.mcp_client

JSON-RPC client for server communication.

Key Classes:

• MCPClient - Communication client
  • start_server() - Start MCP server process
  • send_request() - Send JSON-RPC request
  • stop() - Stop server

Example:

from src.clients.mcp_client import MCPClient

client = MCPClient(server_script="src/server/mcp_server.py")
response = await client.send_request("tools/list")

Adding a New Tool

Step 1: Define Tool in Server

Edit src/server/mcp_server.py:

def _init_tools(self):
    return [
        # ... existing tools ...
        {
            "name": "my_new_tool",
            "description": "What this tool does",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "param1": {"type": "string", "description": "Parameter 1"},
                    "param2": {"type": "number", "description": "Parameter 2"}
                },
                "required": ["param1"]
            }
        }
    ]

Step 2: Implement Tool Handler

In src/server/mcp_server.py:

async def call_tool(self, name: str, arguments: Dict) -> Any:
    if name == "my_new_tool":
        return await self._my_new_tool(arguments)
    # ... other tools ...

async def _my_new_tool(self, args: Dict) -> Dict:
    param1 = args.get("param1")
    param2 = args.get("param2")
    
    # Implementation
    result = f"Processed {param1} with {param2}"
    
    return {
        "status": "success",
        "result": result,
        "timestamp": datetime.now().isoformat()
    }

Step 3: Add UI Handler

In src/ui/components.py, add to ToolCallHandler:

async def my_new_tool(
    self,
    param1: str,
    param2: float
) -> AsyncGenerator[str, None]:
    """Stream my new tool output"""
    if not param1.strip():
        yield " Please enter param1"
        return
    
    yield " Processing...\n"
    
    try:
        response = await self.call_tool(
            "my_new_tool",
            {"param1": param1, "param2": param2}
        )
        
        if response.get("result", {}).get("status") == "success":
            yield response["result"]["result"]
            yield f"\n Complete"
        else:
            yield " Failed"
    except Exception as e:
        yield f" Error: {str(e)}"

Step 4: Add UI Tab

In src/ui/app.py, add to create_app():

with gr.Tab(" My New Tool"):
    with gr.Group(elem_classes="tool-section"):
        gr.Markdown("### Tool Description")
        
        param1 = gr.Textbox(
            label="Parameter 1",
            placeholder="Enter value..."
        )
        param2 = gr.Number(
            label="Parameter 2",
            value=0
        )
        
        btn = gr.Button(" Execute", variant="primary", size="lg")
        output = gr.Markdown(elem_classes="output-box")
        
        btn.click(
            fn=handler.my_new_tool,
            inputs=[param1, param2],
            outputs=output
        )

Development Workflow

1. Create Feature Branch

git checkout -b feature/my-feature

2. Make Changes

• Edit relevant modules in src/
• Follow existing code style
• Add type hints
• Write docstrings

3. Test Locally

python run_ui.py
# Test in browser

4. Commit Changes

git add .
git commit -m "Add my feature"

Code Style Guidelines

Type Hints

def my_function(param1: str, param2: int) -> Dict[str, Any]:
    """Function description"""
    pass

Docstrings

async def my_async_function(name: str) -> AsyncGenerator[str, None]:
    """
    Stream output from server
    
    Args:
        name: The name to process
        
    Yields:
        Streamed output chunks
    """
    pass

Error Handling

try:
    response = await client.send_request(...)
    if response.get("error"):
        yield f" Error: {response['error']}"
    else:
        yield " Success"
except Exception as e:
    yield f" Exception: {str(e)}"

Common Tasks

Debug Server Issues

python run_server.py 2>&1 | grep -i error

Check Server Logs

Server logs are written to stderr, UI logs to stdout

Update Dependencies

pip install -r requirements.txt --upgrade

Run Tests

pytest tests/

Performance Tips

1. Streaming: Use AsyncGenerator for long operations
2. Async/await: Don't block the event loop
3. Error Handling: Catch and handle exceptions gracefully
4. Logging: Use proper logging, not print()

Deployment

Docker

docker build -t ecomcp .
docker run -p 7860:7860 ecomcp

Modal

Update Modal configuration in deployment files

Production Checklist

• Set environment variables
• Configure logging
• Test error handling
• Performance testing
• Security review

Troubleshooting

Issue: Server fails to start

Solution: Check if port 7860 is available

lsof -i :7860

Issue: Timeout errors

Solution: Increase timeout in mcp_client.py or use non-blocking calls

Issue: OpenAI API errors

Solution: Check OPENAI_API_KEY is set and valid

Resources

• Gradio Documentation
• MCP Protocol Spec
• Python Async Guide
• JSON-RPC 2.0 Spec

Getting Help

1. Check existing code for examples
2. Review archived files for reference
3. Check logs for error messages
4. Ask in project issues