771 lines
33 KiB
Python
771 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hmac
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Annotated, Any
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from mcp.server.fastmcp.server import TransportSecuritySettings
|
|
from pydantic import Field
|
|
from starlette.applications import Starlette
|
|
from starlette.routing import Mount, Route
|
|
|
|
from .config import load_config
|
|
from .crm import CrmService
|
|
from .nb_client import NbAuthenticationError, NbClient, NbCommandError
|
|
from .nocobase_http import NocoBaseHttpClient
|
|
from .projects import ProjectService
|
|
from .uploads import UploadEndpoint
|
|
|
|
DEFAULT_HTTP_HOST = "127.0.0.1"
|
|
DEFAULT_HTTP_PORT = 10087
|
|
DEFAULT_HTTP_PATH = "/mcp"
|
|
DEFAULT_TOKEN_ENV = "NOCOBASE_MCP_TOKEN"
|
|
ENABLE_CONFIG_TOOLS_ENV = "NOCOBASE_MCP_ENABLE_CONFIG_TOOLS"
|
|
ENABLE_RAW_DATA_ENV = "NOCOBASE_MCP_ENABLE_RAW_DATA"
|
|
UPLOAD_API_BASE_URL_ENV = "NOCOBASE_MCP_UPLOAD_API_BASE_URL"
|
|
UPLOAD_JWT_ENV = "NOCOBASE_MCP_UPLOAD_JWT"
|
|
UPLOAD_MAX_BYTES_ENV = "NOCOBASE_MCP_UPLOAD_MAX_BYTES"
|
|
API_BASE_URL_ENV = "NOCOBASE_MCP_API_BASE_URL"
|
|
API_TOKEN_ENV = "NOCOBASE_MCP_API_TOKEN"
|
|
DEFAULT_UPLOAD_MAX_BYTES = 25 * 1024 * 1024
|
|
|
|
|
|
def _bearer_token(authorization: str) -> str:
|
|
scheme, _, token = authorization.partition(" ")
|
|
if scheme.lower() != "bearer" or not token:
|
|
return ""
|
|
return token.strip()
|
|
|
|
|
|
def _authorization_header(scope: dict[str, Any]) -> str:
|
|
headers = {key.lower(): value for key, value in scope.get("headers", [])}
|
|
return headers.get(b"authorization", b"").decode("latin1")
|
|
|
|
|
|
class BearerTokenAuthMiddleware:
|
|
def __init__(self, app: Any, token: str) -> None:
|
|
if not token:
|
|
raise ValueError("HTTP MCP token must not be empty")
|
|
self.app = app
|
|
self.token = token
|
|
|
|
async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None:
|
|
if scope.get("type") != "http":
|
|
await self.app(scope, receive, send)
|
|
return
|
|
if str(scope.get("method") or "").upper() == "OPTIONS":
|
|
await send(
|
|
{
|
|
"type": "http.response.start",
|
|
"status": 204,
|
|
"headers": [
|
|
(b"access-control-allow-origin", b"*"),
|
|
(b"access-control-allow-methods", b"POST, OPTIONS"),
|
|
(b"access-control-allow-headers", b"authorization, content-type, mcp-session-id"),
|
|
],
|
|
}
|
|
)
|
|
await send({"type": "http.response.body", "body": b""})
|
|
return
|
|
token = _bearer_token(_authorization_header(scope))
|
|
if not hmac.compare_digest(token, self.token):
|
|
await send(
|
|
{
|
|
"type": "http.response.start",
|
|
"status": 401,
|
|
"headers": [
|
|
(b"content-type", b"application/json"),
|
|
(b"www-authenticate", b"Bearer"),
|
|
],
|
|
}
|
|
)
|
|
await send({"type": "http.response.body", "body": b'{"error":"unauthorized"}'})
|
|
return
|
|
await self.app(scope, receive, send)
|
|
|
|
|
|
def _make_service(config_path: str | Path | None = None) -> CrmService:
|
|
config = load_config(config_path)
|
|
nb_bin = os.environ.get("NOCOBASE_MCP_NB_BIN", "").strip() or "nb"
|
|
client = NbClient(env_name=config.env_name, data_source=config.data_source, nb_bin=nb_bin)
|
|
return CrmService(config, client)
|
|
|
|
|
|
def _make_write_services(
|
|
config_path: str | Path | None = None,
|
|
*,
|
|
api_base_url: str | None = None,
|
|
api_token: str | None = None,
|
|
) -> tuple[CrmService, ProjectService, NocoBaseHttpClient]:
|
|
config = load_config(config_path)
|
|
nb_bin = os.environ.get("NOCOBASE_MCP_NB_BIN", "").strip() or "nb"
|
|
client = NbClient(
|
|
env_name=config.env_name,
|
|
data_source=config.data_source,
|
|
nb_bin=nb_bin,
|
|
api_base_url=api_base_url,
|
|
api_token=api_token,
|
|
)
|
|
if api_base_url and api_token:
|
|
http_client = NocoBaseHttpClient(api_base_url, api_token)
|
|
else:
|
|
http_client = NocoBaseHttpClient(credential_provider=client.environment_credentials)
|
|
return CrmService(config, client), ProjectService(config.project, client, http_client), http_client
|
|
|
|
|
|
def _safe_tool_result(operation: Callable[[], dict[str, Any]]) -> dict[str, Any]:
|
|
try:
|
|
return operation()
|
|
except NbAuthenticationError as error:
|
|
return _tool_error(
|
|
code="nb_auth_failed",
|
|
message=str(error),
|
|
hint="Run `nb env auth <configured-env>` on this computer and complete login with your own NocoBase account.",
|
|
retryable=False,
|
|
)
|
|
except NbCommandError as error:
|
|
return _tool_error(
|
|
code="nb_command_failed",
|
|
message=str(error),
|
|
hint="Check nb env status, NocoBase ACL, collection names, and query parameters.",
|
|
retryable=False,
|
|
)
|
|
except subprocess.TimeoutExpired as error:
|
|
return _tool_error(
|
|
code="nb_command_timeout",
|
|
message=str(error),
|
|
hint="The NocoBase CLI call timed out. Retry with a smaller query or check NocoBase responsiveness.",
|
|
retryable=True,
|
|
)
|
|
except FileNotFoundError as error:
|
|
return _tool_error(
|
|
code="nb_not_found",
|
|
message=str(error),
|
|
hint="Set NOCOBASE_MCP_NB_BIN to the absolute nb path, then restart with scripts/mcp_service.sh start.",
|
|
retryable=False,
|
|
)
|
|
except Exception as error:
|
|
return _tool_error(
|
|
code="tool_error",
|
|
message=str(error),
|
|
hint="Check MCP server logs and the CRM mapping config.",
|
|
retryable=False,
|
|
)
|
|
|
|
|
|
def _safe_raw_tool_result(include_raw: bool, operation: Callable[[], dict[str, Any]]) -> dict[str, Any]:
|
|
if include_raw and os.environ.get(ENABLE_RAW_DATA_ENV, "").strip() != "1":
|
|
return _tool_error(
|
|
code="raw_data_disabled",
|
|
message="Raw NocoBase rows are disabled for this local MCP installation.",
|
|
hint=f"Keep include_raw=false, or set {ENABLE_RAW_DATA_ENV}=1 after approving the data exposure policy.",
|
|
retryable=False,
|
|
)
|
|
return _safe_tool_result(operation)
|
|
|
|
|
|
def _tool_error(*, code: str, message: str, hint: str, retryable: bool) -> dict[str, Any]:
|
|
return {"error": {"code": code, "message": message, "hint": hint, "retryable": retryable}}
|
|
|
|
|
|
def create_server(
|
|
*,
|
|
crm_service: CrmService | None = None,
|
|
project_service: ProjectService | None = None,
|
|
config_path: str | Path | None = None,
|
|
readonly: bool = True,
|
|
host: str = DEFAULT_HTTP_HOST,
|
|
port: int = DEFAULT_HTTP_PORT,
|
|
streamable_http_path: str = DEFAULT_HTTP_PATH,
|
|
allowed_hosts: list[str] | None = None,
|
|
allowed_origins: list[str] | None = None,
|
|
) -> Any:
|
|
if FastMCP is None:
|
|
raise RuntimeError("Missing MCP SDK. Install dependencies with: uv sync")
|
|
service = crm_service or _make_service(config_path)
|
|
if not readonly and project_service is None:
|
|
raise ValueError("project_service is required when readonly is false")
|
|
transport_security = None
|
|
if TransportSecuritySettings is not None and (allowed_hosts is not None or allowed_origins is not None):
|
|
transport_security = TransportSecuritySettings(
|
|
allowed_hosts=allowed_hosts or [],
|
|
allowed_origins=allowed_origins or [],
|
|
)
|
|
server = FastMCP(
|
|
"nocobase-crm",
|
|
host=host,
|
|
port=port,
|
|
streamable_http_path=streamable_http_path,
|
|
transport_security=transport_security,
|
|
)
|
|
|
|
@server.tool()
|
|
def crm_health() -> dict[str, Any]:
|
|
"""Check NocoBase CLI reachability and the configured CRM collection mapping."""
|
|
return _safe_tool_result(service.crm_health)
|
|
|
|
@server.tool()
|
|
def crm_schema(
|
|
include_fields: Annotated[
|
|
bool,
|
|
Field(description="When true, try to include NocoBase collection and field metadata."),
|
|
] = False,
|
|
) -> dict[str, Any]:
|
|
"""Return the configured CRM collection and field mapping for this MCP server."""
|
|
return _safe_tool_result(lambda: service.crm_schema(include_fields=include_fields))
|
|
|
|
@server.tool()
|
|
def crm_overview(
|
|
date_from: Annotated[
|
|
str | None,
|
|
Field(
|
|
description="Optional lower bound for follow-up date, in YYYY-MM-DD or an accepted NocoBase date shortcut."
|
|
),
|
|
] = None,
|
|
date_to: Annotated[
|
|
str | None,
|
|
Field(
|
|
description="Optional upper bound for follow-up date, in YYYY-MM-DD or an accepted NocoBase date shortcut."
|
|
),
|
|
] = None,
|
|
owner: Annotated[
|
|
str | None,
|
|
Field(description="Optional user to filter customer ownership and follow-up participation."),
|
|
] = None,
|
|
) -> dict[str, Any]:
|
|
"""Return CRM totals, common customer distributions, and recent follow-ups."""
|
|
return _safe_tool_result(lambda: service.crm_overview(date_from=date_from, date_to=date_to, owner=owner))
|
|
|
|
@server.tool()
|
|
def crm_filter_options(
|
|
owner_query: Annotated[
|
|
str | None,
|
|
Field(description="Optional owner search text. Returns matching user candidates for owner filters."),
|
|
] = None,
|
|
) -> dict[str, Any]:
|
|
"""Return available customer filter values and optional owner candidates."""
|
|
return _safe_tool_result(lambda: service.crm_filter_options(owner_query=owner_query))
|
|
|
|
@server.tool()
|
|
def resolve_owner(
|
|
query: Annotated[str, Field(description="Owner username, nickname, email, or English name to resolve.")],
|
|
) -> dict[str, Any]:
|
|
"""Return candidate NocoBase users for an owner filter before querying CRM data."""
|
|
return _safe_tool_result(lambda: service.resolve_owner(query))
|
|
|
|
@server.tool()
|
|
def list_teams(
|
|
query: Annotated[str | None, Field(description="Optional team name search.")] = None,
|
|
include_members: Annotated[bool, Field(description="Include resolved team members.")] = False,
|
|
formal_members_only: Annotated[
|
|
bool,
|
|
Field(description="When including members, keep only formal employees."),
|
|
] = True,
|
|
limit: Annotated[int, Field(description="Maximum teams to return.")] = 20,
|
|
offset: Annotated[int, Field(description="Zero-based result offset.")] = 0,
|
|
) -> dict[str, Any]:
|
|
"""List configured CRM teams and optionally their formal members."""
|
|
return _safe_tool_result(
|
|
lambda: service.list_teams(
|
|
query,
|
|
include_members=include_members,
|
|
formal_members_only=formal_members_only,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
)
|
|
|
|
@server.tool()
|
|
def resolve_team(
|
|
query: Annotated[str, Field(description="Team name to resolve.")],
|
|
) -> dict[str, Any]:
|
|
"""Resolve a team name before using it as a CRM filter."""
|
|
return _safe_tool_result(lambda: service.resolve_team(query))
|
|
|
|
@server.tool()
|
|
def get_team(
|
|
team_id: Annotated[Any, Field(description="NocoBase team ID.")],
|
|
include_members: Annotated[bool, Field(description="Include team members.")] = True,
|
|
formal_members_only: Annotated[
|
|
bool,
|
|
Field(description="Return only formal employees when members are included."),
|
|
] = True,
|
|
) -> dict[str, Any]:
|
|
"""Get one CRM team, its managers, and optionally its members."""
|
|
return _safe_tool_result(
|
|
lambda: service.get_team(
|
|
team_id,
|
|
include_members=include_members,
|
|
formal_members_only=formal_members_only,
|
|
)
|
|
)
|
|
|
|
@server.tool()
|
|
def team_crm_overview(
|
|
team: Annotated[str, Field(description="Exact or uniquely matching CRM team name.")],
|
|
date_from: Annotated[str | None, Field(description="Optional lower follow-up date bound.")] = None,
|
|
date_to: Annotated[str | None, Field(description="Optional upper follow-up date bound.")] = None,
|
|
include_member_breakdown: Annotated[
|
|
bool,
|
|
Field(description="Include per-member customer ownership and follow-up contribution."),
|
|
] = True,
|
|
) -> dict[str, Any]:
|
|
"""Return team totals and member contributions using formal CRM team membership."""
|
|
return _safe_tool_result(
|
|
lambda: service.team_crm_overview(
|
|
team,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
include_member_breakdown=include_member_breakdown,
|
|
)
|
|
)
|
|
|
|
if os.environ.get(ENABLE_CONFIG_TOOLS_ENV, "").strip() == "1":
|
|
|
|
@server.tool()
|
|
def crm_config_template() -> dict[str, Any]:
|
|
"""Infer a config.json template from collection field metadata without reading business records."""
|
|
return _safe_tool_result(service.crm_config_template)
|
|
|
|
@server.tool()
|
|
def list_customers(
|
|
query: Annotated[str | None, Field(description="Optional customer keyword search.")] = None,
|
|
status: Annotated[str | None, Field(description="Optional exact customer status.")] = None,
|
|
owner: Annotated[
|
|
str | None, Field(description="Optional customer owner name, username, email, or English name.")
|
|
] = None,
|
|
team: Annotated[str | None, Field(description="Optional CRM team name.")] = None,
|
|
source: Annotated[str | None, Field(description="Optional exact customer source.")] = None,
|
|
limit: Annotated[int, Field(description="Maximum rows to return. Defaults to 20.")] = 20,
|
|
offset: Annotated[int, Field(description="Zero-based result offset. Defaults to 0.")] = 0,
|
|
include_raw: Annotated[
|
|
bool, Field(description="When true, include the original NocoBase row under raw. Defaults to false.")
|
|
] = False,
|
|
) -> dict[str, Any]:
|
|
"""List CRM customers with compact rows and pagination metadata.
|
|
|
|
Rows are {id, name, english_name, territorial_name, status, source,
|
|
owners: [{id, name}], managers, createdAt, updatedAt}. Empty values are
|
|
omitted. A name shown as "#123" means that customer has no name recorded.
|
|
"""
|
|
return _safe_raw_tool_result(
|
|
include_raw,
|
|
lambda: service.list_customers(
|
|
query=query,
|
|
status=status,
|
|
owner=owner,
|
|
team=team,
|
|
source=source,
|
|
limit=limit,
|
|
offset=offset,
|
|
include_raw=include_raw,
|
|
),
|
|
)
|
|
|
|
@server.tool()
|
|
def get_customer(
|
|
customer_id: Annotated[Any, Field(description="Customer primary key value.")],
|
|
include_contacts: Annotated[bool, Field(description="Include related contacts.")] = True,
|
|
include_recent_followups: Annotated[bool, Field(description="Include recent related follow-ups.")] = True,
|
|
followup_limit: Annotated[int, Field(description="Maximum related follow-ups to return.")] = 10,
|
|
include_raw: Annotated[
|
|
bool,
|
|
Field(description="When true, include original NocoBase rows under raw. Defaults to false."),
|
|
] = False,
|
|
) -> dict[str, Any]:
|
|
"""Get one customer with its contacts and recent follow-ups.
|
|
|
|
Returns {customer, contacts, recent_followups, timezone}. The customer is
|
|
{id, name, english_name, territorial_name, status, source, owners, managers,
|
|
address, intro, tags, ...} using readable keys, never NocoBase field codes.
|
|
Contacts are {id, name, title} without phone or email -- call list_contacts
|
|
for those. Follow-ups use the same row shape as list_followups. Empty values
|
|
are omitted, so a missing key means empty rather than unknown.
|
|
"""
|
|
return _safe_raw_tool_result(
|
|
include_raw,
|
|
lambda: service.get_customer(
|
|
customer_id,
|
|
include_contacts=include_contacts,
|
|
include_recent_followups=include_recent_followups,
|
|
followup_limit=followup_limit,
|
|
include_raw=include_raw,
|
|
),
|
|
)
|
|
|
|
@server.tool()
|
|
def list_contacts(
|
|
customer_id: Annotated[
|
|
Any | None, Field(description="Optional customer primary key for relation filtering.")
|
|
] = None,
|
|
query: Annotated[str | None, Field(description="Optional contact keyword search.")] = None,
|
|
limit: Annotated[int, Field(description="Maximum rows to return. Defaults to 20.")] = 20,
|
|
offset: Annotated[int, Field(description="Zero-based result offset. Defaults to 0.")] = 0,
|
|
include_details: Annotated[
|
|
bool,
|
|
Field(description="Include contact phone and email. Defaults to true."),
|
|
] = True,
|
|
include_raw: Annotated[
|
|
bool, Field(description="When true, include the original NocoBase row under raw. Defaults to false.")
|
|
] = False,
|
|
) -> dict[str, Any]:
|
|
"""List CRM contacts as compact rows, optionally filtered by customer.
|
|
|
|
Rows are {id, name, title, phone, email, customers: [{id, name}],
|
|
createdAt, updatedAt}. Set include_details=false to omit phone and email.
|
|
Empty values are omitted.
|
|
"""
|
|
return _safe_raw_tool_result(
|
|
include_raw,
|
|
lambda: service.list_contacts(
|
|
customer_id=customer_id,
|
|
query=query,
|
|
limit=limit,
|
|
offset=offset,
|
|
include_details=include_details,
|
|
include_raw=include_raw,
|
|
),
|
|
)
|
|
|
|
@server.tool()
|
|
def list_followups(
|
|
customer_id: Annotated[
|
|
Any | None, Field(description="Optional customer primary key for relation filtering.")
|
|
] = None,
|
|
contact_id: Annotated[
|
|
Any | None, Field(description="Optional contact primary key for relation filtering.")
|
|
] = None,
|
|
owner: Annotated[
|
|
str | None, Field(description="Optional follow-up participant name, username, email, or English name.")
|
|
] = None,
|
|
team: Annotated[str | None, Field(description="Optional CRM team name.")] = None,
|
|
project_id: Annotated[Any | None, Field(description="Optional CRM project ID.")] = None,
|
|
followup_type: Annotated[str | None, Field(description="Optional exact follow-up type value.")] = None,
|
|
record_status: Annotated[str | None, Field(description="Optional exact follow-up record status value.")] = None,
|
|
date_from: Annotated[str | None, Field(description="Optional lower follow-up date bound.")] = None,
|
|
date_to: Annotated[str | None, Field(description="Optional upper follow-up date bound.")] = None,
|
|
limit: Annotated[int, Field(description="Maximum rows to return. Defaults to 20.")] = 20,
|
|
offset: Annotated[int, Field(description="Zero-based result offset. Defaults to 0.")] = 0,
|
|
include_raw: Annotated[
|
|
bool, Field(description="When true, include the original NocoBase row under raw. Defaults to false.")
|
|
] = False,
|
|
) -> dict[str, Any]:
|
|
"""List customer follow-up records as compact rows with pagination metadata.
|
|
|
|
Rows are {id, followedAt, customers: [{id, name}], contacts, participants,
|
|
projects, type, status, location, content}. Content is the full follow-up
|
|
text, never truncated. type and status are the display labels; use
|
|
crm_filter_options for the codes the followup_type and record_status
|
|
filters expect. creator appears only when the submitter is not also a
|
|
participant. Empty values are omitted.
|
|
"""
|
|
return _safe_raw_tool_result(
|
|
include_raw,
|
|
lambda: service.list_followups(
|
|
customer_id=customer_id,
|
|
contact_id=contact_id,
|
|
owner=owner,
|
|
team=team,
|
|
project_id=project_id,
|
|
followup_type=followup_type,
|
|
record_status=record_status,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
limit=limit,
|
|
offset=offset,
|
|
include_raw=include_raw,
|
|
),
|
|
)
|
|
|
|
@server.tool()
|
|
def list_projects(
|
|
customer_id: Annotated[Any | None, Field(description="Optional customer ID.")] = None,
|
|
team: Annotated[str | None, Field(description="Optional CRM team name.")] = None,
|
|
stage: Annotated[str | None, Field(description="Optional project stage label or configured value.")] = None,
|
|
limit: Annotated[int, Field(description="Maximum projects to return.")] = 20,
|
|
offset: Annotated[int, Field(description="Zero-based result offset.")] = 0,
|
|
include_raw: Annotated[bool, Field(description="Include original NocoBase project rows.")] = False,
|
|
) -> dict[str, Any]:
|
|
"""List CRM projects linked to customers, teams, and project stages."""
|
|
return _safe_raw_tool_result(
|
|
include_raw,
|
|
lambda: service.list_projects(
|
|
customer_id=customer_id,
|
|
team=team,
|
|
stage=stage,
|
|
limit=limit,
|
|
offset=offset,
|
|
include_raw=include_raw,
|
|
),
|
|
)
|
|
|
|
@server.tool()
|
|
def get_project(
|
|
project_id: Annotated[Any, Field(description="NocoBase CRM project ID.")],
|
|
include_contacts: Annotated[bool, Field(description="Include related contacts.")] = True,
|
|
include_followups: Annotated[bool, Field(description="Include related follow-ups.")] = True,
|
|
include_updates: Annotated[bool, Field(description="Include project update records.")] = True,
|
|
) -> dict[str, Any]:
|
|
"""Get a CRM project with its existing NocoBase relations."""
|
|
return _safe_tool_result(
|
|
lambda: service.get_project(
|
|
project_id,
|
|
include_contacts=include_contacts,
|
|
include_followups=include_followups,
|
|
include_updates=include_updates,
|
|
)
|
|
)
|
|
|
|
@server.tool()
|
|
def get_project_timeline(
|
|
project_id: Annotated[Any, Field(description="NocoBase CRM project ID.")],
|
|
date_from: Annotated[str | None, Field(description="Optional lower event date bound.")] = None,
|
|
date_to: Annotated[str | None, Field(description="Optional upper event date bound.")] = None,
|
|
) -> dict[str, Any]:
|
|
"""Return a chronological CRM project timeline from follow-ups and project updates."""
|
|
return _safe_tool_result(lambda: service.get_project_timeline(project_id, date_from=date_from, date_to=date_to))
|
|
|
|
@server.tool()
|
|
def list_stale_customers(
|
|
team: Annotated[str | None, Field(description="Optional CRM team name.")] = None,
|
|
owner: Annotated[str | None, Field(description="Optional customer owner.")] = None,
|
|
inactive_days: Annotated[int, Field(description="Days without a valid follow-up.")] = 30,
|
|
as_of: Annotated[str | None, Field(description="Reference date in YYYY-MM-DD format.")] = None,
|
|
limit: Annotated[int, Field(description="Maximum customers to return.")] = 20,
|
|
offset: Annotated[int, Field(description="Zero-based result offset.")] = 0,
|
|
) -> dict[str, Any]:
|
|
"""List customers whose last valid follow-up is older than the configured threshold."""
|
|
return _safe_tool_result(
|
|
lambda: service.list_stale_customers(
|
|
team=team,
|
|
owner=owner,
|
|
inactive_days=inactive_days,
|
|
as_of=as_of,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
)
|
|
|
|
if not readonly:
|
|
|
|
@server.tool()
|
|
def create_project(
|
|
name: Annotated[str, Field(description="Project name. It must not be blank.")],
|
|
customer_id: Annotated[Any | None, Field(description="Optional NocoBase customer ID.")] = None,
|
|
inquiry_date: Annotated[
|
|
str | None, Field(description="Optional project inquiry date in YYYY-MM-DD format.")
|
|
] = None,
|
|
progress: Annotated[
|
|
str | None, Field(description="Optional configured Chinese project progress label.")
|
|
] = None,
|
|
applications: Annotated[str | None, Field(description="Optional project applications.")] = None,
|
|
part_numbers: Annotated[str | None, Field(description="Optional project part numbers.")] = None,
|
|
period_usage: Annotated[str | None, Field(description="Optional project period usage.")] = None,
|
|
details: Annotated[str | None, Field(description="Optional project details.")] = None,
|
|
attachment_ids: Annotated[
|
|
list[Any] | None, Field(description="Optional uploaded attachment IDs to associate.")
|
|
] = None,
|
|
) -> dict[str, Any]:
|
|
"""Create one fixed-schema NocoBase project and optionally associate uploaded attachments."""
|
|
assert project_service is not None
|
|
return _safe_tool_result(
|
|
lambda: project_service.create_project(
|
|
name=name,
|
|
customer_id=customer_id,
|
|
inquiry_date=inquiry_date,
|
|
progress=progress,
|
|
applications=applications,
|
|
part_numbers=part_numbers,
|
|
period_usage=period_usage,
|
|
details=details,
|
|
attachment_ids=attachment_ids,
|
|
)
|
|
)
|
|
|
|
@server.tool()
|
|
def update_project(
|
|
project_id: Annotated[Any, Field(description="NocoBase project ID.")],
|
|
name: Annotated[str | None, Field(description="Optional project name.")] = None,
|
|
customer_id: Annotated[Any | None, Field(description="Optional NocoBase customer ID.")] = None,
|
|
inquiry_date: Annotated[
|
|
str | None, Field(description="Optional project inquiry date in YYYY-MM-DD format.")
|
|
] = None,
|
|
progress: Annotated[
|
|
str | None, Field(description="Optional configured Chinese project progress label.")
|
|
] = None,
|
|
applications: Annotated[str | None, Field(description="Optional project applications.")] = None,
|
|
part_numbers: Annotated[str | None, Field(description="Optional project part numbers.")] = None,
|
|
period_usage: Annotated[str | None, Field(description="Optional project period usage.")] = None,
|
|
details: Annotated[str | None, Field(description="Optional project details.")] = None,
|
|
add_attachment_ids: Annotated[
|
|
list[Any] | None, Field(description="Attachment IDs to add incrementally.")
|
|
] = None,
|
|
remove_attachment_ids: Annotated[
|
|
list[Any] | None, Field(description="Attachment IDs to remove incrementally.")
|
|
] = None,
|
|
) -> dict[str, Any]:
|
|
"""Update fixed project fields and incrementally add or remove uploaded attachments."""
|
|
assert project_service is not None
|
|
return _safe_tool_result(
|
|
lambda: project_service.update_project(
|
|
project_id,
|
|
name=name,
|
|
customer_id=customer_id,
|
|
inquiry_date=inquiry_date,
|
|
progress=progress,
|
|
applications=applications,
|
|
part_numbers=part_numbers,
|
|
period_usage=period_usage,
|
|
details=details,
|
|
add_attachment_ids=add_attachment_ids,
|
|
remove_attachment_ids=remove_attachment_ids,
|
|
)
|
|
)
|
|
|
|
return server
|
|
|
|
|
|
def _read_token_from_env(env_name: str) -> str:
|
|
token = os.environ.get(env_name, "").strip()
|
|
if not token:
|
|
raise RuntimeError(f"{env_name} must be set for HTTP MCP transport")
|
|
return token
|
|
|
|
|
|
def create_http_app(server: Any, token: str, upload_endpoint: UploadEndpoint | None = None) -> Any:
|
|
if upload_endpoint is None:
|
|
return BearerTokenAuthMiddleware(server.streamable_http_app(), token)
|
|
|
|
async def upload_route(request: Any) -> Any:
|
|
return await upload_endpoint(request)
|
|
|
|
mcp_app = server.streamable_http_app()
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/uploads", endpoint=upload_route, methods=["POST"]),
|
|
Mount("/", app=mcp_app),
|
|
],
|
|
lifespan=mcp_app.router.lifespan_context,
|
|
)
|
|
return BearerTokenAuthMiddleware(app, token)
|
|
|
|
|
|
def _read_nocobase_environment() -> tuple[str | None, str | None, int]:
|
|
canonical_url = os.environ.get(API_BASE_URL_ENV, "").strip().rstrip("/")
|
|
canonical_token = os.environ.get(API_TOKEN_ENV, "").strip()
|
|
legacy_url = os.environ.get(UPLOAD_API_BASE_URL_ENV, "").strip().rstrip("/")
|
|
legacy_token = os.environ.get(UPLOAD_JWT_ENV, "").strip()
|
|
|
|
canonical_any = bool(canonical_url or canonical_token)
|
|
legacy_any = bool(legacy_url or legacy_token)
|
|
if canonical_any and not (canonical_url and canonical_token):
|
|
raise RuntimeError(f"{API_BASE_URL_ENV} and {API_TOKEN_ENV} must be set together")
|
|
if legacy_any and not (legacy_url and legacy_token):
|
|
raise RuntimeError(f"{UPLOAD_API_BASE_URL_ENV} and {UPLOAD_JWT_ENV} must be set together")
|
|
if canonical_any and legacy_any and (canonical_url != legacy_url or canonical_token != legacy_token):
|
|
raise RuntimeError("Canonical and legacy NocoBase credential environment variables conflict")
|
|
if canonical_any:
|
|
api_base_url, api_token = canonical_url, canonical_token
|
|
elif legacy_any:
|
|
api_base_url, api_token = legacy_url, legacy_token
|
|
else:
|
|
api_base_url, api_token = None, None
|
|
|
|
raw_max_bytes = os.environ.get(UPLOAD_MAX_BYTES_ENV, str(DEFAULT_UPLOAD_MAX_BYTES)).strip()
|
|
try:
|
|
max_bytes = int(raw_max_bytes)
|
|
except ValueError as error:
|
|
raise RuntimeError(f"{UPLOAD_MAX_BYTES_ENV} must be a positive integer") from error
|
|
if max_bytes <= 0:
|
|
raise RuntimeError(f"{UPLOAD_MAX_BYTES_ENV} must be a positive integer")
|
|
return api_base_url, api_token, max_bytes
|
|
|
|
|
|
def run_http_server(
|
|
*,
|
|
config_path: Path | None,
|
|
host: str,
|
|
port: int,
|
|
token_env: str,
|
|
streamable_http_path: str,
|
|
allowed_hosts: list[str] | None,
|
|
allowed_origins: list[str] | None,
|
|
) -> None:
|
|
import uvicorn
|
|
|
|
token = _read_token_from_env(token_env)
|
|
api_base_url, api_token, max_bytes = _read_nocobase_environment()
|
|
crm_service, project_service, http_client = _make_write_services(
|
|
config_path,
|
|
api_base_url=api_base_url,
|
|
api_token=api_token,
|
|
)
|
|
server = create_server(
|
|
crm_service=crm_service,
|
|
project_service=project_service,
|
|
readonly=False,
|
|
host=host,
|
|
port=port,
|
|
streamable_http_path=streamable_http_path,
|
|
allowed_hosts=allowed_hosts,
|
|
allowed_origins=allowed_origins,
|
|
)
|
|
upload_endpoint = UploadEndpoint(http_client, max_bytes=max_bytes, temp_dir=Path(tempfile.gettempdir()))
|
|
app = create_http_app(server, token, upload_endpoint=upload_endpoint)
|
|
uvicorn.run(app, host=host, port=port, log_level="info")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run the local NocoBase CRM MCP server.")
|
|
parser.add_argument("--config", default=None, type=Path, help="CRM mapping JSON config path.")
|
|
parser.add_argument(
|
|
"--transport",
|
|
choices=["stdio", "streamable-http"],
|
|
default="stdio",
|
|
help="MCP transport. HTTP mode requires a bearer token.",
|
|
)
|
|
parser.add_argument("--host", default=DEFAULT_HTTP_HOST, help="HTTP bind host.")
|
|
parser.add_argument("--port", default=DEFAULT_HTTP_PORT, type=int, help="HTTP port.")
|
|
parser.add_argument("--path", default=DEFAULT_HTTP_PATH, help="Streamable HTTP MCP path.")
|
|
parser.add_argument(
|
|
"--token-env", default=DEFAULT_TOKEN_ENV, help="Environment variable containing the HTTP bearer token."
|
|
)
|
|
parser.add_argument("--allowed-host", action="append", dest="allowed_hosts")
|
|
parser.add_argument("--allowed-origin", action="append", dest="allowed_origins")
|
|
parser.add_argument(
|
|
"--enable-writes",
|
|
action="store_true",
|
|
help="Expose fixed-schema project create/update tools in local stdio mode.",
|
|
)
|
|
args = parser.parse_args()
|
|
if args.transport == "stdio":
|
|
if args.enable_writes:
|
|
crm_service, project_service, _http_client = _make_write_services(args.config)
|
|
create_server(
|
|
crm_service=crm_service,
|
|
project_service=project_service,
|
|
readonly=False,
|
|
).run(transport="stdio")
|
|
else:
|
|
create_server(config_path=args.config).run(transport="stdio")
|
|
return
|
|
run_http_server(
|
|
config_path=args.config,
|
|
host=args.host,
|
|
port=args.port,
|
|
token_env=args.token_env,
|
|
streamable_http_path=args.path,
|
|
allowed_hosts=args.allowed_hosts,
|
|
allowed_origins=args.allowed_origins,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|