Customising tools for your hosted Patchworks MCP server

Introduction

The hosted Patchworks MCP server includes a set of default tools that are ready to use. You can't edit these, but you can add your own.

To do this, you create a draft server deployment and then write Python directly in the editor. When the draft is deployed, your custom tools become available alongside the default set.

You need a working knowledge of Python, Pydantic, the FastMCP framework, and the Patchworks API to add MCP tools.

Need to know

• Tools can only be added or updated in a draft server deployment. You can create a draft from an existing deployed server or from an inactive (i.e. previously deployed) version.

• Any new tools must follow a defined pattern.

• Code for built-in tools is not exposed - you cannot edit or remove default tools.

Default template

When you create a new MCP server, a default template is provided:

    # Import required types inside the function
    from typing import Any, Optional, Dict
    from mcp.server.fastmcp import Context
    from pydantic import BaseModel, Field
    import os, logging
    from urllib.parse import quote
    import requests

    # Set up logging
    log = logging.getLogger("patchworks-client")
    log.setLevel(logging.INFO)
    log.info("Logging setup complete")
    # ------------------------------------------------------------------------------
    # Patchworks MCP Server - Built-in Tools
    # ------------------------------------------------------------------------------
    # There are already 15 tools provided by Patchworks for use out of the box:
    #
    # Flow Management:
    #   - get_all_flows: List flows from the Core API
    #   - get_flow_runs: Query flow runs (filter by status, started_after; sort; includes)
    #   - get_flow_run_logs: Retrieve logs for a specific flow run (optionally with payload IDs)
    #   - start_flow: Trigger a flow run via the Start service (/flows/{id}/start)
    #
    # Troubleshooting:
    #   - summarise_failed_run: Summarise what went wrong in a failed run
    #   - triage_latest_failures: Fetch recent failed runs and return a compact summary
    #   - download_payload: Download payload bytes for a given payload metadata ID (base64)
    #
    # Data Pools:
    #   - list_data_pools: List all data/dedupe pools
    #   - get_deduped_data: Retrieve deduplicated data for a specific pool
    #
    # Marketplace:
    #   - get_marketplace_apps: List marketplace apps from the Patchworks marketplace
    #   - get_marketplace_app: Get details of a specific marketplace app by ID
    #
    # Flow Creation:
    #   - create_process_flow_from_prompt: Build a flow from natural-language prompt
    #   - create_process_flow_from_json: Import a flow with the exact JSON body provided
    #
    # Utility:
    #   - example_get_all_flows: Example demonstrating tool definition patterns
    #   - list_tools: List all available tools in this MCP server
    # ------------------------------------------------------------------------------


    # ------------------------------------------------------------------------------
    # Add Your Custom Tools Below
    # ------------------------------------------------------------------------------

    # Available in the environment:
    # - patchworks_client as pw (for defined Patchworks API calls shown above)
    # - get_token_from_context(ctx) (for authentication)
    # - @mcp.tool() decorator
    # - Context, FastMCP from mcp.server.fastmcp
    # - BaseModel, Field from pydantic

    # Pattern for adding a new tool:
    #
    # Define your argument model
    # --------------------------------------
    # class MyCustomToolArgs(BaseModel):
    #     param1: str = Field(..., description="Description of param1")
    #
    #
    # Define the tool
    # --------------------------------------
    # @mcp.tool()
    # def my_custom_tool(args: MyCustomToolArgs, ctx: Optional[Context] = None):
    #     """Description of what your tool does."""
    #     token = get_token_from_context(ctx)
    #     result = pw.some_api_method(args.param1, token=token)
    #     return result
    #
    #
    # Register the tool with the MCP client
    # --------------------------------------
    # mcp._custom_tools_registry.append({
    #     "name": "my_custom_tool",
    #     "description": "Description of what your tool does.",
    #     "category": "custom"
    # })
    def _core_get(path: str, token: str, params: Optional[Dict[str, Any]] = None) -> Any:
        """
        Minimal Core API GET helper for custom tools.
        Assumes CORE_BASE_URL is available in env or already defined in this file.
        """
        CORE_API = os.getenv("PATCHWORKS_CORE_API", "").rstrip("/")
        url = f"{CORE_API}{path}"

        headers = {
            "Authorization": f"{token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
        r = requests.get(url, headers=headers, params=params, timeout=30)

        # Raise a helpful error (FastMCP will surface this)
        try:
            r.raise_for_status()
        except requests.HTTPError as e:
            # include response body if present (often has useful error JSON)
            raise RuntimeError(f"Core API GET {url} failed: {r.status_code} {r.text} {headers.Authorization}") from e

        # Return JSON if possible, else raw text
        content_type = (r.headers.get("Content-Type") or "").lower()
        if "application/json" in content_type:
            return r.json()
        return r.text


    # ------------------------------------------------------------------------------
    # Example tool - Get all Caches
    # ------------------------------------------------------------------------------
    class ListCachesArgs(BaseModel):
        page: int = Field(1, ge=1, description="Page number to retrieve")
        per_page: int = Field(50, ge=1, le=200, description="Number of items per page")
        include: Optional[str] = Field(None, description="Comma-separated includes (optional)")


    @mcp.tool()
    def list_caches(args: ListCachesArgs, ctx: Optional[Context] = None) -> Any:
        """List caches from the Core API."""

        token = get_token_from_context(ctx)
        params: Dict[str, Any] = {"page": args.page, "per_page": args.per_page}
        if args.include:
            params["include"] = args.include
        return _core_get("/caches", token=token, params=params)


    mcp._custom_tools_registry.append({
        "name": "list_caches",
        "description": "List caches from the Core API.",
        "category": "caches"
    })

We break this file down in the following sections:

• Import libraries

• Set up logging

• Built-in tools reference

• Environment

• Tool pattern

• Core helper

• Example tool

• Where do I add the next tool?

Import libraries

Loads required libraries, including type hints, Pydantic for argument validation, the MCP context, HTTP requests, and logging utilities.

Lines 1-7 from main example

    # Import required types inside the function
    from typing import Any, Optional, Dict
    from mcp.server.fastmcp import Context
    from pydantic import BaseModel, Field
    import os, logging
    from urllib.parse import quote
    import requests

Set up logging

Configures a logger so tools can output diagnostic messages during execution.

Lines 9-12 from main example

    # Set up logging
    log = logging.getLogger("patchworks-client")
    log.setLevel(logging.INFO)
    log.info("Logging setup complete")

Built-in tools reference

A commented list of the built-in tools. This is just reference information, so you can see what's already available.

Lines 13-44 from main example

    # ------------------------------------------------------------------------------
    # Patchworks MCP Server - Built-in Tools
    # ------------------------------------------------------------------------------
    # There are already 15 tools provided by Patchworks for use out of the box:
    #
    # Flow Management:
    #   - get_all_flows: List flows from the Core API
    #   - get_flow_runs: Query flow runs (filter by status, started_after; sort; includes)
    #   - get_flow_run_logs: Retrieve logs for a specific flow run (optionally with payload IDs)
    #   - start_flow: Trigger a flow run via the Start service (/flows/{id}/start)
    #
    # Troubleshooting:
    #   - summarise_failed_run: Summarise what went wrong in a failed run
    #   - triage_latest_failures: Fetch recent failed runs and return a compact summary
    #   - download_payload: Download payload bytes for a given payload metadata ID (base64)
    #
    # Data Pools:
    #   - list_data_pools: List all data/dedupe pools
    #   - get_deduped_data: Retrieve deduplicated data for a specific pool
    #
    # Marketplace:
    #   - get_marketplace_apps: List marketplace apps from the Patchworks marketplace
    #   - get_marketplace_app: Get details of a specific marketplace app by ID
    #
    # Flow Creation:
    #   - create_process_flow_from_prompt: Build a flow from a natural-language prompt
    #   - create_process_flow_from_json: Import a flow with the exact JSON body provided
    #
    # Utility:
    #   - example_get_all_flows: Example demonstrating tool definition patterns
    #   - list_tools: List all available tools in this MCP server
    # ------------------------------------------------------------------------------

Environment

Lines 51-56 from main example

    # Available in the environment:
    # - patchworks_client as pw (for defined Patchworks API calls shown above)
    # - get_token_from_context(ctx) (for authentication)
    # - @mcp.tool() decorator
    # - Context, FastMCP from mcp.server.fastmcp
    # - BaseModel, Field from pydantic

A commented summary explaining what's available in the environment (pw, get_token_from_context, decorators, etc.).

Tool pattern

A commented template showing the required three-part pattern for adding a tool: argument model, tool definition, and registry:

Lines 58-82 from main example

    # Pattern for adding a new tool:
    #
    # Define your argument model
    # --------------------------------------
    # class MyCustomToolArgs(BaseModel):
    #     param1: str = Field(..., description="Description of param1")
    #
    #
    # Define the tool
    # --------------------------------------
    # @mcp.tool()
    # def my_custom_tool(args: MyCustomToolArgs, ctx: Optional[Context] = None):
    #     """Description of what your tool does."""
    #     token = get_token_from_context(ctx)
    #     result = pw.some_api_method(args.param1, token=token)
    #     return result
    #
    #
    # Register the tool with the MCP client
    # --------------------------------------
    # mcp._custom_tools_registry.append({
    #     "name": "my_custom_tool",
    #     "description": "Description of what your tool does.",
    #     "category": "custom"
    # })

Use this pattern for any tools that you want to add.

Core helper

A reusable function that handles making authenticated requests to the Patchworks Core API. It reads the API base URL from the environment, attaches auth headers, makes the request, handles errors, and returns the JSON response. This helper is used by every tool.

Lines 83-109 from main example

   def _core_get(path: str, token: str, params: Optional[Dict[str, Any]] = None) -> Any:
        """
        Minimal Core API GET helper for custom tools.
        Assumes CORE_BASE_URL is available in env or already defined in this file.
        """
        CORE_API = os.getenv("PATCHWORKS_CORE_API", "").rstrip("/")
        url = f"{CORE_API}{path}"

        headers = {
            "Authorization": f"{token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
        r = requests.get(url, headers=headers, params=params, timeout=30)

        # Raise a helpful error (FastMCP will surface this)
        try:
            r.raise_for_status()
        except requests.HTTPError as e:
            # include response body if present (often has useful error JSON)
            raise RuntimeError(f"Core API GET {url} failed: {r.status_code} {r.text} {headers.Authorization}") from e

        # Return JSON if possible, else raw text
        content_type = (r.headers.get("Content-Type") or "").lower()
        if "application/json" in content_type:
            return r.json()
        return r.text

Example tool (get all caches)

A working example (get all caches) showing the three-part pattern in practice.

Lines 112-136 from main example

    # ------------------------------------------------------------------------------
    # Example tool - Get all Caches
    # ------------------------------------------------------------------------------
    class ListCachesArgs(BaseModel):
        page: int = Field(1, ge=1, description="Page number to retrieve")
        per_page: int = Field(50, ge=1, le=200, description="Number of items per page")
        include: Optional[str] = Field(None, description="Comma-separated includes (optional)")


    @mcp.tool()
    def list_caches(args: ListCachesArgs, ctx: Optional[Context] = None) -> Any:
        """List caches from the Core API."""

        token = get_token_from_context(ctx)
        params: Dict[str, Any] = {"page": args.page, "per_page": args.per_page}
        if args.include:
            params["include"] = args.include
        return _core_get("/caches", token=token, params=params)


    mcp._custom_tools_registry.append({
        "name": "list_caches",
        "description": "List caches from the Core API.",
        "category": "caches"
    })

Where do I add the next tool?

To extend the default specification, add your next tool right after the last. See line 138 onwards in the full example below, where we've added a tool to get connectors:

    # Import required types inside the function
    from typing import Any, Optional, Dict
    from mcp.server.fastmcp import Context
    from pydantic import BaseModel, Field
    import os, logging
    from urllib.parse import quote
    import requests

    # Set up logging
    log = logging.getLogger("patchworks-client")
    log.setLevel(logging.INFO)
    log.info("Logging setup complete")
    # ------------------------------------------------------------------------------
    # Patchworks MCP Server - Built-in Tools
    # ------------------------------------------------------------------------------
    # There are already 15 tools provided by Patchworks for use out of the box:
    #
    # Flow Management:
    #   - get_all_flows: List flows from the Core API
    #   - get_flow_runs: Query flow runs (filter by status, started_after; sort; includes)
    #   - get_flow_run_logs: Retrieve logs for a specific flow run (optionally with payload IDs)
    #   - start_flow: Trigger a flow run via the Start service (/flows/{id}/start)
    #
    # Troubleshooting:
    #   - summarise_failed_run: Summarise what went wrong in a failed run
    #   - triage_latest_failures: Fetch recent failed runs and return a compact summary
    #   - download_payload: Download payload bytes for a given payload metadata ID (base64)
    #
    # Data Pools:
    #   - list_data_pools: List all data/dedupe pools
    #   - get_deduped_data: Retrieve deduplicated data for a specific pool
    #
    # Marketplace:
    #   - get_marketplace_apps: List marketplace apps from the Patchworks marketplace
    #   - get_marketplace_app: Get details of a specific marketplace app by ID
    #
    # Flow Creation:
    #   - create_process_flow_from_prompt: Build a flow from natural-language prompt
    #   - create_process_flow_from_json: Import a flow with the exact JSON body provided
    #
    # Utility:
    #   - example_get_all_flows: Example demonstrating tool definition patterns
    #   - list_tools: List all available tools in this MCP server
    # ------------------------------------------------------------------------------


    # ------------------------------------------------------------------------------
    # Add Your Custom Tools Below
    # ------------------------------------------------------------------------------

    # Available in the environment:
    # - patchworks_client as pw (for defined Patchworks API calls shown above)
    # - get_token_from_context(ctx) (for authentication)
    # - @mcp.tool() decorator
    # - Context, FastMCP from mcp.server.fastmcp
    # - BaseModel, Field from pydantic

    # Pattern for adding a new tool:
    #
    # Define your argument model
    # --------------------------------------
    # class MyCustomToolArgs(BaseModel):
    #     param1: str = Field(..., description="Description of param1")
    #
    #
    # Define the tool
    # --------------------------------------
    # @mcp.tool()
    # def my_custom_tool(args: MyCustomToolArgs, ctx: Optional[Context] = None):
    #     """Description of what your tool does."""
    #     token = get_token_from_context(ctx)
    #     result = pw.some_api_method(args.param1, token=token)
    #     return result
    #
    #
    # Register the tool with the MCP client
    # --------------------------------------
    # mcp._custom_tools_registry.append({
    #     "name": "my_custom_tool",
    #     "description": "Description of what your tool does.",
    #     "category": "custom"
    # })
    def _core_get(path: str, token: str, params: Optional[Dict[str, Any]] = None) -> Any:
        """
        Minimal Core API GET helper for custom tools.
        Assumes CORE_BASE_URL is available in env or already defined in this file.
        """
        CORE_API = os.getenv("PATCHWORKS_CORE_API", "").rstrip("/")
        url = f"{CORE_API}{path}"

        headers = {
            "Authorization": f"{token}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }
        r = requests.get(url, headers=headers, params=params, timeout=30)

        # Raise a helpful error (FastMCP will surface this)
        try:
            r.raise_for_status()
        except requests.HTTPError as e:
            # include response body if present (often has useful error JSON)
            raise RuntimeError(f"Core API GET {url} failed: {r.status_code} {r.text} {headers.Authorization}") from e

        # Return JSON if possible, else raw text
        content_type = (r.headers.get("Content-Type") or "").lower()
        if "application/json" in content_type:
            return r.json()
        return r.text


    # ------------------------------------------------------------------------------
    # Example tool - Get all Caches
    # ------------------------------------------------------------------------------
    class ListCachesArgs(BaseModel):
        page: int = Field(1, ge=1, description="Page number to retrieve")
        per_page: int = Field(50, ge=1, le=200, description="Number of items per page")
        include: Optional[str] = Field(None, description="Comma-separated includes (optional)")


    @mcp.tool()
    def list_caches(args: ListCachesArgs, ctx: Optional[Context] = None) -> Any:
        """List caches from the Core API."""

        token = get_token_from_context(ctx)
        params: Dict[str, Any] = {"page": args.page, "per_page": args.per_page}
        if args.include:
            params["include"] = args.include
        return _core_get("/caches", token=token, params=params)


    mcp._custom_tools_registry.append({
        "name": "list_caches",
        "description": "List caches from the Core API.",
        "category": "caches"
    })

    # ------------------------------------------------------------------------------
    # Example tool - Get all Connectors with Instances
    # ------------------------------------------------------------------------------
    class ListConnectorsArgs(BaseModel):
        page: int = Field(1, ge=1, description="Page number to retrieve")
        per_page: int = Field(50, ge=1, le=200, description="Number of items per page")
        include: Optional[str] = Field("instances", description="Comma-separated includes (default: instances)")


    @mcp.tool()
    def list_connectors(args: ListConnectorsArgs, ctx: Optional[Context] = None) -> Any:
        """List all connectors and their associated instances from the Core API."""

        token = get_token_from_context(ctx)
        params: Dict[str, Any] = {"page": args.page, "per_page": args.per_page}
        if args.include:
            params["include"] = args.include
        return _core_get("/connectors", token=token, params=params)


    mcp._custom_tools_registry.append({
        "name": "list_connectors",
        "description": "List all connectors and their associated instances from the Core API.",
        "category": "connectors"
    })

Editing an MCP server draft deployment

As noted previously, to add your own tools to the Patchworks MCP server, you must work with a draft deployment.

1

Access your MCP server deployments

Select the settings option (at the bottom of the navigation menu):

Then select the MCP servers option:

2

Create a draft deployment (if not already present)

If your server deployments list already includes a draft entry, you can skip this step. Otherwise, you can create a draft using ONE of the methods below:

If a deployed or inactive entry IS present:

Create a draft from it

This creates a new draft entry, leaving any existing deployed version in place (so your AI assistant(s) can continue to access the hosted Patchworks MCP server while you make changes).

Click the ellipsis icon for the required entry:

Then select the create draft from option:

Undeploy the current deployment

If a deployed entry is present, you can undeploy it. This changes the version from deployed to draft, ready for editing. However, it's important to note that your AI assistant(s) cannot access the hosted Patchworks MCP server when it is undeployed.

Click the ellipsis icon for the required entry:

Then select the undeploy option:

If a deployed or inactive entry IS NOT present:

Create a new server deployment, then undeploy it

Follow instructions to deploy a hosted Patchworks MCP server for the first time, then immediately undeploy it (see above) to change the deployed status to draft).

If a deployed entry IS present, but you want to start again from scratch:

Create a new server deployment, then undeploy it

1. Follow instructions to undeploy the deployed version, setting it back to draft.

2. Follow instructions to delete the draft entry.

3. Follow instructions to deploy a hosted Patchworks MCP server for the first time, then immediately undeploy it to change the deployed status to draft).

3

Edit the draft entry

Click the ellipsis icon for the draft entry, then select the edit option. For example:

4

Make required updates

Use the code editor to add/adjust tools:

5

Save or save & deploy the draft

If you want to save your changes but not deploy them yet, click the save button:

Your changes are saved and the draft status does not change.

Keep in mind that you can't test your changes until the draft is deployed.

If you're ready to deploy the draft version, click the save and deploy button:

The draft status changes to deployed. If there was already a deployed version, this becomes inactive.

Once your changes are deployed, you can test the updated tools via the MCP servers page and/or your AI assistant.

Deploying an updated draft does not affect existing connections with AI assistants - these will pick up the new/changed tools when requests are made.