#
tokens: 1024/50000 5/5 files
lines: off (toggle) GitHub
raw markdown copy
# Directory Structure

```
├── .gitignore
├── .python-version
├── assets
│   ├── kickoff.jpg
│   ├── status.jpg
│   └── tools.jpg
├── crewai_enterprise_server.py
├── pyproject.toml
├── README.md
└── uv.lock
```

# Files

--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------

```
3.12

```

--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------

```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Environment variables
.env

```

--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------

```markdown
# CrewAI Enterprise MCP Server

## Overview

A Model Context Protocol (MCP) server implementation that provides deployed CrewAI workflows. This server enables kicking off your deployed crew and inspect the status giving the results of your crew.

## Tools

- kickoff_crew
- get_crew_status

## Env Variables

retrieve from app.crewai.com
`MCP_CREWAI_ENTERPRISE_SERVER_URL`
`MCP_CREWAI_ENTERPRISE_BEARER_TOKEN`

# Usage with Claude Desktop

![Claude Desktop with CrewAI Enterprise MCP Server](assets/tools.jpg)
![Claude Desktop with CrewAI Enterprise MCP Server](assets/kickoff.jpg)
![Claude Desktop with CrewAI Enterprise MCP Server](assets/status.jpg)

To use this MCP server with Claude Desktop, follow these steps:

1. Open Claude Desktop
2. Go to Settings > Developer Settings
3. Add a new MCP server with the configuration shown below

## Locally, cloned repo:

Install `mcp` and `mcp[cli]` locally

```json
{
  "mcpServers": {
    "crewai_enterprise_server": {
      "command": "uv",
      "args": [
        "run",
        "--with",
        "mcp[cli]",
        "mcp",
        "run",
        "<filepath of cloned repo>",
        "/crewai_enterprise_server.py"
      ],
      "env": {
        "MCP_CREWAI_ENTERPRISE_SERVER_URL": "<>",
        "MCP_CREWAI_ENTERPRISE_BEARER_TOKEN": "<>"
      }
    }
  }
}
```

## TODO: Added on PyPI:

```

--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------

```toml
[project]
name = "crew-enterprise-mcp-server-app"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "httpx>=0.28.1",
    "mcp[cli]>=1.4.1",
]

```

--------------------------------------------------------------------------------
/crewai_enterprise_server.py:
--------------------------------------------------------------------------------

```python
import dotenv
import os
import httpx
from mcp.server.fastmcp import FastMCP
from typing import Any


dotenv.load_dotenv()
mcp = FastMCP("crewai_enterprise_server")

CREWAI_ENTERPRISE_SERVER_URL = os.getenv("MCP_CREWAI_ENTERPRISE_SERVER_URL")
CREWAI_ENTERPRISE_BEARER_TOKEN = os.getenv("MCP_CREWAI_ENTERPRISE_BEARER_TOKEN")


@mcp.tool()
async def kickoff_crew(inputs: dict[str, Any]) -> dict[str, Any]:
    """Start a new crew task

    Args:
        inputs: Dictionary containing the query and other input parameters

    Returns:
        Dictionary containing the crew task response. The response will contain the crew id which needs to be returned to check the status of the crew.
    """
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CREWAI_ENTERPRISE_SERVER_URL}/kickoff",
            headers={
                "Authorization": f"Bearer {CREWAI_ENTERPRISE_BEARER_TOKEN}",
                "Content-Type": "application/json",
            },
            json={"inputs": inputs},
        )
        response_json = response.json()
        return response_json


@mcp.tool()
async def get_crew_status(crew_id: str) -> dict[str, Any]:
    """Get the status of a crew task

    Args:
        crew_id: The ID of the crew task to check

    Returns:
        Dictionary containing the crew task status
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{CREWAI_ENTERPRISE_SERVER_URL}/status/{crew_id}",
            headers={
                "Authorization": f"Bearer {CREWAI_ENTERPRISE_BEARER_TOKEN}",
                "Content-Type": "application/json",
            },
        )
        return response.json()


if __name__ == "__main__":
    import mcp

    mcp.run()

```