from __future__ import annotations import json from datetime import date from mcp.server.fastmcp import FastMCP from .backend_client import BackendClient from .config import DEFAULT_PROFILE from .response_trim import USER_SERVER_METHOD_MAP, trim_error_response, trim_public_response, wrap_trimmed_methods from .session_store import SessionStore from .tools.app_tools import AppTools from .tools.auth_tools import AuthTools from .tools.code_block_tools import CodeBlockTools from .tools.directory_tools import DirectoryTools from .tools.export_tools import ExportTools from .tools.feedback_tools import FeedbackTools from .tools.file_tools import FileTools from .tools.import_tools import ImportTools from .tools.resource_read_tools import ResourceReadTools from .tools.task_context_tools import TaskContextTools from .tools.workspace_tools import WorkspaceTools def build_user_server() -> FastMCP: today = date.today() current_year = today.year server = FastMCP( "Qingflow App User MCP", instructions=f"""Use this server for Qingflow operational workflows. Current date: `{today.isoformat()}`. ## App Discovery If `app_key` is unknown, use `app_list` first. Pass `query` to filter visible apps by keyword. If the app is known but the data range is not, use `app_get` first and choose from `accessible_views`. Treat an explicit `view_id` as the exact frontend view context. If `system:all` fails, do not silently switch to `system:initiated`, `system:todo`, or another system view unless the user, frontend URL, or `app_get.accessible_views` explicitly selects that view. If an accessible view has `analysis_supported=false`, do not use it for `record_access` or `record_list`. `boardView` and `ganttView` are special UI views, not data-access targets. `view_get(view_id=...)` also returns `export_capability`; it only means there is a supported export route, not that export permission has been verified. ## Shared Helper `feedback_submit` is always available as a cross-cutting helper. - Use it when the current MCP capability is unsupported, awkward, or still cannot satisfy the user's need after reasonable use. - It does not require Qingflow login or workspace selection. - Call it only after the user explicitly confirms submission. ## Schema-First Rule Call `record_insert_schema_get` before `record_insert`. For simple field changes after the target record is clear, call `record_update` directly. Use `record_update_schema_get` for diagnostics, ambiguous fields, or complex writable-scope inspection. Prefer `record_code_block_schema_get` before `record_code_block_run` when field selection or binding diagnostics are unclear; if the exact code-block field id is already known from record/task detail, run directly. Call `app_get` first when the data range is unclear, then use `record_browse_schema_get(view_id=...)` before `record_access`, `record_list`, `record_get`, or `record_logs_get`. Call `record_import_schema_get` when the import field mapping is unclear before template download or verify. - All `field_id` values must come from the schema response. - Never guess field names or ids. ## Schema Scope `record_insert_schema_get` returns the current user's insert-ready applicant schema; read `required_fields`, `optional_fields`, `runtime_linked_required_fields`, and `payload_template`. Inside `optional_fields`, any field with `may_become_required=true` is still writable, but may become required when linked visibility or option-driven runtime rules activate. `record_update_schema_get` returns the current record's update-ready writable field set and route diagnostics. When `view_id` is explicit, it is the exact frontend view context and must not be silently replaced by another view. `record_browse_schema_get(view_id=...)` returns the same readable fields shown in the selected Qingflow table view header. `record_access.fields` / CSV columns and `record_list.columns / where / order_by / query_fields` use that exact same view schema; a missing field means it is not readable in that view. `searchQueIds` is a backend full-text search scope, not an output-column/projection mechanism. `record_code_block_schema_get` returns code-block-ready schema for exact code block field selection when the applicant schema is readable. `record_import_schema_get` returns import-ready column metadata. - Hidden fields are omitted. - Missing fields mean the field is not visible in the current permission scope. - Read the top-level schema payload directly; do not guess missing writable fields. ## Analytics Path `app_get -> record_browse_schema_get(view_id=...) -> record_access -> Python` Prefer `view_id` entries from `accessible_views` where `analysis_supported=true`. Use `record_access` to write local CSV shard files, then use Python to compute counts, rankings, ratios, trends, and final conclusions. `record_access` does not return bulk `items`; read `files[].local_path`. CSV columns are readable and field-id anchored, such as `项目状态__field_343283094`, and `fields[]` is the compact metadata source. For analysis-style tasks, prefer an explicit time range or business filter. If `record_access.status == "needs_scope"`, do not treat it as a failure; ask for a time/business scope or retry with a user-provided period using `scope.suggested_time_fields` / `scope.recommended_where_examples`. If `record_access.status == "partial"`, read the returned files only as a limited subset and do not give a final full-population conclusion. Use `chart_get` only when the user provides a report URL / chart_id or explicitly asks to read an existing report. Do not use QingBI as the default analysis route. Use this data-access DSL shape: - `columns`: `[{{field_id}}]` - `where`: `[{{field_id, op, value}}]` - `order_by`: `[{{field_id, direction}}]` Important key rules: - Use `op` - Do **not** use `type` - Do **not** use `agg` - Do **not** use `aggregation` - Do **not** use `operator` `record_list` is for browsing and sample checks, not final analysis conclusions. For fuzzy single-record lookup, use `record_list(query=..., query_fields=[{{field_id}}])` to find candidates, read `lookup.next_action`, and only call `record_get` after one candidate is clear. `record_list.query_fields` maps to backend full-text search scope (`searchQueIds`); `record_list.columns` only controls displayed fields. Analysis answers must include concrete numbers. When applicable, include percentages based on the returned totals. ## Record CRUD Path `app_get -> record_browse_schema_get(view_id=...) -> record_list / record_get / record_logs_get` `record_insert_schema_get -> record_insert(items)` `record_update` for simple updates; `record_update_schema_get -> record_update` when the writable field scope is unclear. `record_list / record_get -> record_delete(system view_id)` `record_code_block_run` directly when the exact code-block field is known; otherwise `record_code_block_schema_get -> record_code_block_run` `portal_list -> portal_get -> chart_get / view_get` `portal_get -> view_get -> record_list` - Use `columns` as `[{{field_id}}]` - Use `query` plus optional `query_fields` when the user provides fuzzy record-identifying text - Use `where` items as `{{field_id, op, value}}` - Use `order_by` items as `{{field_id, direction}}` - Legacy forms such as bare integer `field_id`, `fieldId`, `operator`, `values`, or `order` may still parse, but they are compatibility-only and not the canonical DSL - `record_insert` defaults to an applicant-node `items` array; each item contains a field-title keyed `fields` map. A single insert is one item. - `record_update` uses a field-title keyed `fields` map. It first tries the data-manager direct update route, then falls back to the frontend custom-view detail edit route when the selected view can cover the payload; if a unique current-user todo task for the same record exposes editable fields, it can finally use the workflow save-only route. On success, read `status`, `update_route`, and `verification_status`; on failure, read the failure reason and route diagnostics. - For insert, `runtime_linked_required_fields` means required-but-not-directly-writable fields that are usually supplied by runtime linkage or upstream context. - For insert, fields marked `may_become_required=true` stay in `optional_fields`; they are still directly writable, but linked visibility or option-driven rules can make them required at runtime. - Read field-level `linkage` whenever present on `record_insert_schema_get` or `record_update_schema_get`; it is the static hint for linked visibility, reference-driven auto fill, and formula/default auto-fill behavior. - `linkage.sources` lists upstream field titles that influence the current field; `linkage.affects_fields` lists downstream fields that may change when the current field changes. - `linkage.kind=logic_visibility` means linked visibility or option-driven rules are involved; `linkage.kind=reference_fill` means reference/default matching logic is involved; `linkage.kind=formula_fill` means formula/default auto-fill logic is involved. - `record_update_schema_get` exposes the writable field set and update route candidates for the record; with an explicit `view_id`, diagnostics stay scoped to that selected frontend view. Not every field combination is guaranteed; `record_update` still needs data-manager permission, one single matched custom view that can cover the payload, or one unique editable current-user todo task. - `record_delete` deletes by `record_id` or `record_ids`, and requires an accessible system `view_id`; use custom views only to locate records, not as the delete route. - `record_get` is the single-record frontend detail context tool. It returns detail-page visible fields, one-level relation targets, first-page data/workflow logs, associated views/reports, local readable image assets, local downloadable file assets, unavailable context, and `semantic_context`. - Use `record_logs_get` only when the user needs the full visible data/workflow log history for a specific record. It writes JSONL files locally and returns file paths plus completeness metadata; do not expect full log arrays in the response. - Read record images from `record_get.media_assets.items[].local_path` when `readable_by_agent=true`; read attachments/documents/tables from `record_get.file_assets.items[].local_path` and `extraction.text_path` when present. `record_get` follows the frontend storage cookie redirect path for Qingflow attachments, and remote file URLs should not be treated as directly readable. - `record_get.columns` are focus hints only; they do not project the detail fields. Read facts from top-level `fields[]`. - When readback shape matters after insert or update, prefer `record_get` for human/detail-page context, or `record_list(..., output_profile="normalized")` for batch-shaped normalized rows. - Read relation targets from `record_insert_schema_get` / `record_update_schema_get` relation metadata before preparing relation writes. - Member, department, and relation fields may be written with natural strings directly on `record_insert` / `record_update`; only fall back to candidate tools when the user wants explicit candidate browsing or the write returns ambiguity that needs confirmation. - CLI-only agents can use `qingflow record member-candidates --app-key APP_KEY --field-id FIELD_ID --keyword NAME --json` or `qingflow record department-candidates --app-key APP_KEY --field-id FIELD_ID --keyword DEPT --json` for that fallback. - For batch insert `partial_success`, read `created_record_ids`, failed `items[].row_number`, and `failed_fields`; repair only failed rows and never retry the whole batch after any row has `write_executed=true`. - When candidate browsing must match a real update/write scope, pass `record_id`, `workflow_node_id`, and any pending `fields` context to the candidate tool; otherwise the candidate result is only a static applicant-node preview. - If explicit candidate browsing is needed for default-all member or department fields, prefer those field candidate tools instead of starting with `directory_*`. ## Code Block Path Use `record_code_block_run` when the user wants to execute a form code-block field against an existing record. - Prefer resolving the exact code-block field from `record_code_block_schema_get`, but do not treat applicant-schema 40002 as final denial when record/task detail already exposes the code-block field id. - `record_code_block_run` uses the record/task detail answers for the execution context; when applicant schema is unavailable it can still execute with a numeric code-block field id, but schema-bound relation writeback may be skipped. - Treat code-block execution as write-capable, not read-only. - If the code block is bound to relation outputs, Qingflow may calculate target answers and write them back automatically. - For safe debugging, pass `apply_writeback=false` and inspect the parsed alias results plus `relation.calculated_answers_preview` before allowing any writeback. - In workflow context, pass `role=3` and the exact `workflow_node_id`. - After execution, inspect `outputs.configured_aliases`, `outputs.alias_results`, `outputs.alias_map`, `relation.target_fields`, and `writeback.verification` before claiming success. ## Import Path `app_get -> record_import_schema_get -> record_import_template_get -> record_import_verify -> (optional authorized record_import_repair_local) -> record_import_start -> record_import_status_get` - Check `app_get.data.import_capability` before doing import work. - If `import_capability.can_import=false`, stop before template download, file repair, or import start. - Import must go through `verify -> start`; do not start directly from a raw file path. - `record_import_start` requires an explicit `being_enter_auditing` choice. Do not assume a default. - Do not modify user-uploaded files unless the user explicitly authorizes repair. - If repair is authorized, keep the original file and repair a copy, then run `record_import_verify` again before `record_import_start`. ## Export Path Use export only when the user explicitly asks to export/download/generate an Excel or export file. `view_get -> record_export_start -> record_export_status_get -> record_export_get` - `record_export_direct` is the one-shot path that starts export, waits, downloads locally, and still returns remote download links. - Do not use `record_export_direct` as the default analysis path; use `record_access -> Python` instead. - Export v1 supports record views only and follows the same public `view_id` semantics as `record_list`. - `record_export_start` / `record_export_direct` support frontend-like row selection: - omit `record_ids` to export all rows in the selected view - pass `record_ids` to export selected rows only - `record_export_start` / `record_export_direct` also support internal query selection: - pass `where` to resolve matching `record_id` values first - pass `order_by` to keep the internal query and export row order aligned with `record_list` - then run native export as selected rows - `where/order_by` and `record_ids` are mutually exclusive - `record_export_start` / `record_export_direct` also support frontend-like column selection: - omit `columns` to export all current-view fields - pass `columns` to export only selected fields, preserving the provided order - `include_workflow_log=true` maps to the native workflow-log export switch. ## Task Workflow Path `task_list -> task_get -> task_action_execute` - `task_list` returns task-card summaries keyed by `task_id`. - For detail reads and actions, pass the exact `task_id` from `task_list.data.items[].task_id`. - `task_id` is not a row number, list index, record id, or workflow node id. - `task_action_execute` only uses `task_id`; do not reconstruct or pass `app_key + record_id + workflow_node_id` for actions. - `task_workflow_log_get(task_id=...)` and `task_associated_report_detail_get(task_id=...)` are also supported for the current todo context. - Use `task_associated_report_detail_get` for associated view or report details. - Use `task_workflow_log_get` for the current task context workflow log page. For full record-level data/workflow logs, first choose an accessible view with `app_get`, then call `record_logs_get(app_key, record_id, view_id)` with that same explicit `view_id`. - Treat `task_action_execute` as the tool-level action enum surface; the current task's real actions are only the ones listed in `task_get.capabilities.available_actions`. - Use `task_action_execute(action="save_only", fields=...)` when the user wants to save editable field changes on the current node without advancing the workflow. - `save_only` is exposed only when the backend current-node `editableQueIds` signal returns a non-empty result; MCP no longer infers `save_only` from local schema reconstruction. ## Time Handling Normalize relative dates before building DSL. - If the user says `3月` without a year, use the current year: `{current_year}` - Convert month-only phrases into explicit legal date ranges - Never send impossible dates such as `2026-02-29` ## Environment Default to `prod` unless the user explicitly specifies `test`. ## Constraints Avoid builder-side app or schema changes here. ## Feedback Path If the current MCP capability is unsupported, the workflow is awkward, or the user's need still cannot be satisfied after reasonable use, offer to submit product feedback. - First summarize what is still not working - Ask the user whether to submit feedback - Call `feedback_submit` only after explicit user confirmation""", ) sessions = SessionStore() backend = BackendClient() auth = wrap_trimmed_methods(AuthTools(sessions, backend), USER_SERVER_METHOD_MAP) apps = wrap_trimmed_methods(AppTools(sessions, backend), USER_SERVER_METHOD_MAP) workspace = wrap_trimmed_methods(WorkspaceTools(sessions, backend), USER_SERVER_METHOD_MAP) file_tools = wrap_trimmed_methods(FileTools(sessions, backend), USER_SERVER_METHOD_MAP) imports = wrap_trimmed_methods(ImportTools(sessions, backend), USER_SERVER_METHOD_MAP) exports = wrap_trimmed_methods(ExportTools(sessions, backend), USER_SERVER_METHOD_MAP) resources = wrap_trimmed_methods(ResourceReadTools(sessions, backend), USER_SERVER_METHOD_MAP) feedback = FeedbackTools(backend, mcp_side="App User MCP") code_block_tools = wrap_trimmed_methods(CodeBlockTools(sessions, backend), USER_SERVER_METHOD_MAP) task_context_tools = wrap_trimmed_methods(TaskContextTools(sessions, backend), USER_SERVER_METHOD_MAP) directory_tools = wrap_trimmed_methods(DirectoryTools(sessions, backend), USER_SERVER_METHOD_MAP) @server.tool() def auth_use_credential( profile: str = DEFAULT_PROFILE, base_url: str | None = None, qf_version: str | None = None, credential: str = "", persist: bool = False, ) -> dict: return auth.auth_use_credential( profile=profile, base_url=base_url, qf_version=qf_version, credential=credential, persist=persist, ) @server.tool() def auth_whoami(profile: str = DEFAULT_PROFILE) -> dict: return auth.auth_whoami(profile=profile) @server.tool() def auth_logout(profile: str = DEFAULT_PROFILE, forget_persisted: bool = False) -> dict: return auth.auth_logout(profile=profile, forget_persisted=forget_persisted) @server.tool() def record_export_start( profile: str = DEFAULT_PROFILE, app_key: str = "", view_id: str = "", columns: list[dict | int] | None = None, where: list[dict] | None = None, order_by: list[dict] | None = None, record_id: str | int | None = None, record_ids: list[str | int] | None = None, include_workflow_log: bool = False, ) -> dict: return exports.record_export_start( profile=profile, app_key=app_key, view_id=view_id, columns=columns or [], where=where or [], order_by=order_by or [], record_id=record_id, record_ids=record_ids or [], include_workflow_log=include_workflow_log, ) @server.tool() def record_export_status_get( profile: str = DEFAULT_PROFILE, export_handle: str = "", ) -> dict: return exports.record_export_status_get(profile=profile, export_handle=export_handle) @server.tool() def record_export_get( profile: str = DEFAULT_PROFILE, export_handle: str = "", download_to_path: str | None = None, ) -> dict: return exports.record_export_get( profile=profile, export_handle=export_handle, download_to_path=download_to_path, ) @server.tool() def record_export_direct( profile: str = DEFAULT_PROFILE, app_key: str = "", view_id: str = "", columns: list[dict | int] | None = None, where: list[dict] | None = None, order_by: list[dict] | None = None, record_id: str | int | None = None, record_ids: list[str | int] | None = None, include_workflow_log: bool = False, download_to_path: str | None = None, wait_timeout_seconds: float | None = None, ) -> dict: return exports.record_export_direct( profile=profile, app_key=app_key, view_id=view_id, columns=columns or [], where=where or [], order_by=order_by or [], record_id=record_id, record_ids=record_ids or [], include_workflow_log=include_workflow_log, download_to_path=download_to_path, wait_timeout_seconds=wait_timeout_seconds, ) @server.tool() def workspace_list( profile: str = DEFAULT_PROFILE, page_num: int = 1, page_size: int = 20, include_external: bool = False, ) -> dict: return workspace.workspace_list( profile=profile, page_num=page_num, page_size=page_size, include_external=include_external, ) @server.tool() def workspace_get( profile: str = DEFAULT_PROFILE, ws_id: int | None = None, ) -> dict: return workspace.workspace_get( profile=profile, ws_id=ws_id, ) @server.tool() def workspace_select( profile: str = DEFAULT_PROFILE, ws_id: int = 0, ) -> dict: return workspace.workspace_select( profile=profile, ws_id=ws_id, ) @server.tool() def app_list(profile: str = DEFAULT_PROFILE, query: str = "", keyword: str = "") -> dict: return apps.app_list(profile=profile, query=query, keyword=keyword) @server.tool() def app_get(profile: str = DEFAULT_PROFILE, app_key: str = "") -> dict: return apps.app_get(profile=profile, app_key=app_key) @server.tool() def portal_list(profile: str = DEFAULT_PROFILE) -> dict: return resources.portal_list(profile=profile) @server.tool() def portal_get(profile: str = DEFAULT_PROFILE, dash_key: str = "") -> dict: return resources.portal_get(profile=profile, dash_key=dash_key) @server.tool() def view_get(profile: str = DEFAULT_PROFILE, view_id: str = "") -> dict: return resources.view_get(profile=profile, view_id=view_id) @server.tool() def chart_get(profile: str = DEFAULT_PROFILE, chart_id: str = "") -> dict: return resources.chart_get(profile=profile, chart_id=chart_id) @server.tool() def file_get_upload_info( profile: str = DEFAULT_PROFILE, upload_kind: str = "attachment", file_name: str = "", file_size: int = 0, upload_mark: str | None = None, content_type: str | None = None, bucket_type: str | None = None, path_id: int | None = None, file_related_url: str | None = None, ) -> dict: return file_tools.file_get_upload_info( profile=profile, upload_kind=upload_kind, file_name=file_name, file_size=file_size, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) @server.tool() def file_upload_local( profile: str = DEFAULT_PROFILE, upload_kind: str = "attachment", file_path: str = "", upload_mark: str | None = None, content_type: str | None = None, bucket_type: str | None = None, path_id: int | None = None, file_related_url: str | None = None, ) -> dict: return file_tools.file_upload_local( profile=profile, upload_kind=upload_kind, file_path=file_path, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) imports.register(server) @server.tool() def feedback_submit( category: str = "", title: str = "", description: str = "", expected_behavior: str | None = None, actual_behavior: str | None = None, impact_scope: str | None = None, tool_name: str | None = None, app_key: str | None = None, record_id: str | int | None = None, workflow_node_id: str | int | None = None, note: str | None = None, ) -> dict: try: return trim_public_response( "user:feedback_submit", feedback.feedback_submit( category=category, title=title, description=description, expected_behavior=expected_behavior, actual_behavior=actual_behavior, impact_scope=impact_scope, tool_name=tool_name, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, note=note, ), ) except RuntimeError as exc: try: payload = json.loads(str(exc)) except json.JSONDecodeError: raise if isinstance(payload, dict): raise RuntimeError(json.dumps(trim_error_response(payload), ensure_ascii=False)) from None raise code_block_tools.register(server) task_context_tools.register(server) directory_tools.register_frontend_search(server) return server mcp = build_user_server() def main() -> None: mcp.run() if __name__ == "__main__": main()