# 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:
--------------------------------------------------------------------------------
```
1 | 3.12
2 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 |
12 | # Environment variables
13 | .env
14 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # CrewAI Enterprise MCP Server
2 |
3 | ## Overview
4 |
5 | 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.
6 |
7 | ## Tools
8 |
9 | - kickoff_crew
10 | - get_crew_status
11 |
12 | ## Env Variables
13 |
14 | retrieve from app.crewai.com
15 | `MCP_CREWAI_ENTERPRISE_SERVER_URL`
16 | `MCP_CREWAI_ENTERPRISE_BEARER_TOKEN`
17 |
18 | # Usage with Claude Desktop
19 |
20 | 
21 | 
22 | 
23 |
24 | To use this MCP server with Claude Desktop, follow these steps:
25 |
26 | 1. Open Claude Desktop
27 | 2. Go to Settings > Developer Settings
28 | 3. Add a new MCP server with the configuration shown below
29 |
30 | ## Locally, cloned repo:
31 |
32 | Install `mcp` and `mcp[cli]` locally
33 |
34 | ```json
35 | {
36 | "mcpServers": {
37 | "crewai_enterprise_server": {
38 | "command": "uv",
39 | "args": [
40 | "run",
41 | "--with",
42 | "mcp[cli]",
43 | "mcp",
44 | "run",
45 | "<filepath of cloned repo>",
46 | "/crewai_enterprise_server.py"
47 | ],
48 | "env": {
49 | "MCP_CREWAI_ENTERPRISE_SERVER_URL": "<>",
50 | "MCP_CREWAI_ENTERPRISE_BEARER_TOKEN": "<>"
51 | }
52 | }
53 | }
54 | }
55 | ```
56 |
57 | ## TODO: Added on PyPI:
58 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "crew-enterprise-mcp-server-app"
3 | version = "0.1.0"
4 | description = "Add your description here"
5 | readme = "README.md"
6 | requires-python = ">=3.12"
7 | dependencies = [
8 | "httpx>=0.28.1",
9 | "mcp[cli]>=1.4.1",
10 | ]
11 |
```
--------------------------------------------------------------------------------
/crewai_enterprise_server.py:
--------------------------------------------------------------------------------
```python
1 | import dotenv
2 | import os
3 | import httpx
4 | from mcp.server.fastmcp import FastMCP
5 | from typing import Any
6 |
7 |
8 | dotenv.load_dotenv()
9 | mcp = FastMCP("crewai_enterprise_server")
10 |
11 | CREWAI_ENTERPRISE_SERVER_URL = os.getenv("MCP_CREWAI_ENTERPRISE_SERVER_URL")
12 | CREWAI_ENTERPRISE_BEARER_TOKEN = os.getenv("MCP_CREWAI_ENTERPRISE_BEARER_TOKEN")
13 |
14 |
15 | @mcp.tool()
16 | async def kickoff_crew(inputs: dict[str, Any]) -> dict[str, Any]:
17 | """Start a new crew task
18 |
19 | Args:
20 | inputs: Dictionary containing the query and other input parameters
21 |
22 | Returns:
23 | 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.
24 | """
25 | async with httpx.AsyncClient() as client:
26 | response = await client.post(
27 | f"{CREWAI_ENTERPRISE_SERVER_URL}/kickoff",
28 | headers={
29 | "Authorization": f"Bearer {CREWAI_ENTERPRISE_BEARER_TOKEN}",
30 | "Content-Type": "application/json",
31 | },
32 | json={"inputs": inputs},
33 | )
34 | response_json = response.json()
35 | return response_json
36 |
37 |
38 | @mcp.tool()
39 | async def get_crew_status(crew_id: str) -> dict[str, Any]:
40 | """Get the status of a crew task
41 |
42 | Args:
43 | crew_id: The ID of the crew task to check
44 |
45 | Returns:
46 | Dictionary containing the crew task status
47 | """
48 | async with httpx.AsyncClient() as client:
49 | response = await client.get(
50 | f"{CREWAI_ENTERPRISE_SERVER_URL}/status/{crew_id}",
51 | headers={
52 | "Authorization": f"Bearer {CREWAI_ENTERPRISE_BEARER_TOKEN}",
53 | "Content-Type": "application/json",
54 | },
55 | )
56 | return response.json()
57 |
58 |
59 | if __name__ == "__main__":
60 | import mcp
61 |
62 | mcp.run()
63 |
```