Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/workspace.py:10-45 (handler)

  The main handler function for the 'get_workspaces' tool, decorated with @mcp.tool which registers it. Fetches workspaces using the Prefect client and handles local vs Cloud differences.

  @mcp.tool
  async def get_workspaces(
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      name: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get a list of accessible workspaces.
      
      Args:
          limit: Maximum number of workspaces to return
          offset: Number of workspaces to skip
          name: Filter workspaces by name (note: filtering may not be supported by all Prefect versions)
          
      Returns:
          A list of workspaces with their details
      """
      try:
          async with get_client() as client:
              workspaces = await client.read_workspaces(
                  limit=limit,
                  offset=offset,
              )
              
              workspaces_result = {
                  "workspaces": [workspace.dict() for workspace in workspaces]
              }
              
              return [types.TextContent(type="text", text=str(workspaces_result))]
      except Exception as e:
          # For local Prefect instances, workspace APIs may not be available
          return [types.TextContent(
              type="text",
              text="Workspaces are only available in Prefect Cloud. This appears to be a local Prefect instance."
          )]