# Clone and install
git clone https://github.com/openmcp-pro/openmcp.git
cd openmcp
pip install -e .

# Create configuration and get your API key
openmcp init-config

# Start the server (FastMCP with streamable-http transport)
openmcp serve

# Default: Official MCP with streamable-http transport (recommended)
openmcp serve

# Official MCP with SSE transport (real-time streaming)
openmcp serve --transport sse

# Legacy REST API (for custom integrations)
openmcp serve --transport rest --port 9000

import asyncio
import openmcp

async def main():
    # Super simple - just one line!
    mcp = openmcp.MCP("browseruse")
    
    # Create a browser session
    session = await mcp.create_session()
    
    # Navigate and take screenshot
    await session.navigate("https://example.com")
    await session.screenshot("example.png")
    
    # Clean up
    await session.close()

# Or even simpler with context manager:
async def even_simpler():
    async with openmcp.browser() as browser:
        await browser.navigate("https://example.com")
        await browser.click("#some-button")
        await browser.screenshot("result.png")

# One-liner screenshot
async def one_liner():
    await openmcp.screenshot("https://example.com", "quick.png")

asyncio.run(main())

# From localhost - no auth needed!
import httpx

async def localhost_example():
    # Direct API calls without authentication
    async with httpx.AsyncClient() as client:
        response = await client.get("http://localhost:9000/api/v1/services")
        print(response.json())

# Or with the simple interface
import openmcp
mcp = openmcp.MCP("browseruse")  # No API key needed from localhost!

# Works from anywhere - great for CI/CD
import openmcp

# Using mock API key
mcp = openmcp.MCP("browseruse", api_key="openmcp-localhost-auth")

# Or direct HTTP with mock key
headers = {"Authorization": "Bearer openmcp-localhost-auth"}

# Generate production API key
openmcp init-config  # Shows your API key

# Use in production
export OPENMCP_API_KEY="bmcp_your-production-key-here"

# config.yaml
auth:
  allow_localhost: true           # Enable localhost bypass (default: true)
  mock_api_key: "openmcp-localhost-auth"  # Mock key for testing
  secret_key: "your-secret-key"   # JWT secret

import openmcp

# Just one line to get started!
mcp = openmcp.MCP("browseruse")

mcp = openmcp.MCP("browseruse")
session = await mcp.create_session()
await session.navigate("https://example.com")
await session.click("#button")
await session.screenshot("result.png")
await session.close()

async with openmcp.browser() as browser:
    await browser.navigate("https://example.com")
    await browser.type("#search", "openmcp")
    await browser.click("#submit")
    await browser.screenshot("search_results.png")
# Automatically cleaned up!

# Quick screenshot
await openmcp.screenshot("https://example.com", "page.png")

# Quick form test
await openmcp.test_form("https://example.com/contact", {
    "#name": "John Doe",
    "#email": "john@example.com"
})

import openmcp

# Cursor will generate code like this:
async def test_website():
    async with openmcp.browser() as browser:
        await browser.navigate("https://yoursite.com")
        await browser.click("#login-button")
        await browser.type("#email", "test@example.com")
        await browser.screenshot("login_page.png")

# cursor_openmcp.py - Add this to your project
import httpx
import asyncio
from typing import Dict, Any, Optional

class CursorWebAutomation:
    """Web automation helper for Cursor AI assistant."""
    
    def __init__(self, api_key: str = "", base_url: str = "http://localhost:9000"):
        self.api_key = api_key
        self.base_url = base_url
        # Only add auth header if API key is provided (localhost mode doesn't need it)
        self.headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
        self.session_id: Optional[str] = None
    
    async def start_browser(self, headless: bool = True) -> str:
        """Start a browser session for automation."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/services/browseruse/call",
                headers=self.headers,
                json={
                    "tool_name": "create_session",
                    "arguments": {"headless": headless, "timeout": 30}
                }
            )
            result = response.json()
            if result["success"]:
                self.session_id = result["result"]["session_id"]
                return self.session_id
            raise Exception(f"Failed to start browser: {result}")
    
    async def navigate(self, url: str) -> Dict[str, Any]:
        """Navigate to a URL."""
        if not self.session_id:
            await self.start_browser()
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/services/browseruse/call",
                headers=self.headers,
                json={
                    "tool_name": "navigate",
                    "arguments": {"url": url},
                    "session_id": self.session_id
                }
            )
            return response.json()
    
    async def find_and_click(self, selector: str) -> Dict[str, Any]:
        """Find and click an element."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/services/browseruse/call",
                headers=self.headers,
                json={
                    "tool_name": "click_element",
                    "arguments": {"selector": selector},
                    "session_id": self.session_id
                }
            )
            return response.json()
    
    async def type_text(self, selector: str, text: str) -> Dict[str, Any]:
        """Type text into an input field."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/services/browseruse/call",
                headers=self.headers,
                json={
                    "tool_name": "type_text",
                    "arguments": {"selector": selector, "text": text},
                    "session_id": self.session_id
                }
            )
            return response.json()
    
    async def screenshot(self, save_path: str = "screenshot.png") -> str:
        """Take a screenshot and save it."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/v1/services/browseruse/call",
                headers=self.headers,
                json={
                    "tool_name": "take_screenshot",
                    "arguments": {},
                    "session_id": self.session_id
                }
            )
            result = response.json()
            if result["success"]:
                import base64
                screenshot_data = base64.b64decode(result["result"]["screenshot"])
                with open(save_path, "wb") as f:
                    f.write(screenshot_data)
                return save_path
            raise Exception("Screenshot failed")

# Usage example for Cursor
async def demo_for_cursor():
    """Demo function that Cursor can call."""
    # No API key needed from localhost! For remote use: api_key="openmcp-localhost-auth"
    automation = CursorWebAutomation()  # Empty for localhost, or use mock key
    
    # Navigate to a website
    await automation.navigate("https://example.com")
    
    # Take a screenshot
    screenshot_path = await automation.screenshot("example_page.png")
    print(f"Screenshot saved: {screenshot_path}")
    
    # You can now ask Cursor to:
    # - Automate form filling
    # - Scrape website data  
    # - Test web applications
    # - Generate web automation scripts

# Run this in Cursor's terminal
if __name__ == "__main__":
    asyncio.run(demo_for_cursor())

// .cursor/mcp_config.json  
{
  "mcpServers": {
    "openmcp": {
      "command": "openmcp",
      "args": ["serve", "--transport", "sse"],
      "env": {}
    }
  }
}

// Claude Desktop MCP Configuration
{
  "mcpServers": {
    "openmcp-browser": {
      "command": "openmcp", 
      "args": ["serve", "--transport", "sse"]
    }
  }
}

You: "Please help me test the contact form on https://example.com"

Claude: I'll help you test the contact form. Let me:
1. Create a browser session
2. Navigate to the website  
3. Find the contact form
4. Fill it out with test data
5. Take screenshots of the process

[Claude automatically uses openmcp tools to complete the task]

# Start with streamable-http transport
openmcp serve
# or explicitly:
openmcp serve --transport streamable-http

from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession

async with streamablehttp_client('http://localhost:8000/') as (read_stream, write_stream, session_id):
    session = ClientSession(read_stream, write_stream)
    async with session:
        await session.initialize()
        tools = await session.list_tools()

# Start with SSE transport
openmcp serve --transport sse

from mcp.client.sse import sse_client
from mcp.client.session import ClientSession

async with sse_client('http://localhost:8000/sse') as (read_stream, write_stream):
    session = ClientSession(read_stream, write_stream)
    async with session:
        await session.initialize()
        tools = await session.list_tools()

# Start with REST transport
openmcp serve --transport rest --port 9000

import httpx

# Direct REST API calls
async with httpx.AsyncClient() as client:
    response = await client.get("http://localhost:9000/api/v1/services")
    services = response.json()
    
    # Create browser session
    response = await client.post(
        "http://localhost:9000/api/v1/services/browseruse/call",
        json={"tool_name": "create_session", "arguments": {"headless": True}}
    )

# test_automation.py - Generated by Cursor with openmcp
import asyncio
from cursor_openmcp import CursorWebAutomation

async def test_login_flow():
    """Test user login functionality."""
    automation = CursorWebAutomation()  # No auth needed from localhost!
    
    # Navigate to login page
    await automation.navigate("https://yourapp.com/login")
    
    # Fill login form
    await automation.type_text("#email", "test@example.com")
    await automation.type_text("#password", "testpass123")
    
    # Submit form
    await automation.find_and_click("#login-button")
    
    # Verify success
    await asyncio.sleep(2)  # Wait for redirect
    screenshot = await automation.screenshot("login_success.png")
    
    print(f"Login test completed. Screenshot: {screenshot}")

# Cursor can generate and run this automatically!
asyncio.run(test_login_flow())

You: "Please collect all product names and prices from https://shop.example.com"

Claude: I'll help you scrape the product data. Let me:

1. Navigate to the shop website
2. Find all product elements  
3. Extract names and prices
4. Save the data to a CSV file

[Uses openmcp tools automatically]

Here's the collected data:
- Product A: $29.99
- Product B: $39.99
- Product C: $19.99

Data saved to products.csv

# Generated by AI assistant using openmcp
async def automate_job_application():
    automation = CursorWebAutomation()  # Localhost = no auth required!
    
    # Navigate to job application
    await automation.navigate("https://company.com/careers/apply")
    
    # Fill personal information
    await automation.type_text("#first-name", "John")
    await automation.type_text("#last-name", "Doe")
    await automation.type_text("#email", "john@example.com")
    
    # Upload resume (if file upload is supported)
    await automation.find_and_click("#resume-upload")
    
    # Take screenshot for verification
    await automation.screenshot("application_filled.png")
    
    print("Job application automated successfully!")

# config.yaml
server:
  host: "0.0.0.0"
  port: 9000
  debug: false

auth:
  secret_key: "your-secret-key"
  
services:
  - name: "browseruse"
    enabled: true
    config:
      headless: true
      timeout: 30
      max_sessions: 10  # Allow more concurrent sessions
      
  # Add more services here
  - name: "future_service"
    enabled: false

# .env
OPENMCP_SECRET_KEY=your-production-secret
OPENMCP_HOST=0.0.0.0
OPENMCP_PORT=9000
OPENMCP_MAX_SESSIONS=20

# Build and run
docker build -t openmcp .
docker run -p 9000:9000 -e OPENMCP_SECRET_KEY=your-secret openmcp

# docker-compose.yml
version: '3.8'
services:
  openmcp:
    build: .
    ports:
      - "9000:9000"
    environment:
      - OPENMCP_SECRET_KEY=${SECRET_KEY}
    volumes:
      - ./config.yaml:/app/config.yaml
    restart: unless-stopped

# Generate secure API keys
openmcp create-key "production-agent" --expires 365

# Rotate keys regularly
openmcp create-key "new-key" --expires 90
# Update your applications
# Revoke old key

# config.yaml - Production settings
server:
  host: "127.0.0.1"  # Only local access
  port: 9000

auth:
  allow_localhost: false  # Disable localhost bypass in production
  mock_api_key: ""        # Disable mock key in production

# Use reverse proxy (nginx/traefik) for HTTPS
# Enable firewall rules
# Use VPN for remote access

# Monitor server health
curl http://localhost:9000/health

# Check service status
curl -H "Authorization: Bearer $API_KEY" \
     http://localhost:9000/api/v1/services/browseruse/status

# Structured logging is built-in
import structlog
logger = structlog.get_logger()

# All API calls are automatically logged
# Browser actions are tracked
# Errors are captured with context

# Input commit message:
"Add new feature
Claude helped with implementation  
Fix authentication bug"

# Automatically filtered to:
"Add new feature
Fix authentication bug"

import asyncio
import httpx
import json

async def stream_browser_session():
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            "http://localhost:9000/api/v1/services/browseruse/stream",
            json={
                "tool_name": "create_session",
                "arguments": {"headless": True}
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    print(f"{event['type']}: {event['message']}")
                    
                    if event['type'] == 'progress':
                        print(f"Progress: {event.get('progress', 0)}%")

asyncio.run(stream_browser_session())

# Python streaming examples
python examples/simple_sse_demo.py
python examples/sse_streaming_example.py

# Web interface
open examples/sse_web_demo.html

POST /api/v1/services/{service_name}/stream
Content-Type: application/json

{
  "tool_name": "navigate",
  "arguments": {"url": "https://example.com"},
  "session_id": "session-id-here"
}

# my_custom_service.py
from openmcp.services.base import BaseMCPService

class MyCustomService(BaseMCPService):
    def get_tools(self):
        return [{
            "name": "my_tool",
            "description": "My custom automation tool",
            "parameters": {
                "type": "object",
                "properties": {
                    "input": {"type": "string"}
                }
            }
        }]
    
    async def call_tool(self, tool_name, arguments, session_id=None):
        if tool_name == "my_tool":
            # Implement your custom logic
            return {"result": f"Processed: {arguments['input']}"}

# Check port availability
netstat -an | grep 9000

# Try different port
openmcp serve --port 8001

# Install Chrome/Chromium
# Ubuntu/Debian:
sudo apt-get install chromium-browser

# macOS:
brew install --cask google-chrome

# Windows: Download from Google

# From localhost? No auth needed!
curl http://localhost:9000/api/v1/services

# Use mock key for testing
curl -H "Authorization: Bearer openmcp-localhost-auth" \
     http://localhost:9000/api/v1/services

# Regenerate API key for production
openmcp init-config --force

# Check key format (should start with 'bmcp_')
echo $API_KEY

# Verify MCP server starts with correct transport
openmcp serve --transport sse

# Check MCP configuration syntax in your client
# Restart Cursor/Claude Desktop after configuration changes

# Test official MCP client connection
python -c "
from mcp.client.sse import sse_client
import asyncio

async def test():
    async with sse_client('http://localhost:8000/sse') as streams:
        print('✅ MCP connection successful')

asyncio.run(test())
"